diff --git a/common/arg.cpp b/common/arg.cpp index a287b907d..975344131 100644 --- a/common/arg.cpp +++ b/common/arg.cpp @@ -850,8 +850,9 @@ static bool common_params_parse_ex(int argc, char ** argv, common_params_context params.kv_overrides.back().key[0] = 0; } - if (!params.server_tools.empty() && !params.cors_origins_explicit) { - LOG_WRN("server tools are enabled, using localhost as default CORS origin (change via --cors-origins)\n"); + const bool mcp_enabled = !params.mcp_servers_config.empty() || !params.mcp_servers_json.empty(); + if ((!params.server_tools.empty() || mcp_enabled) && !params.cors_origins_explicit) { + LOG_WRN("server tools or MCP servers are enabled, using localhost as default CORS origin (change via --cors-origins)\n"); params.cors_origins = "localhost"; } @@ -3261,6 +3262,22 @@ common_params_context common_params_parser_init(common_params & params, llama_ex params.server_tools = parse_csv_row(value); } ).set_examples({LLAMA_EXAMPLE_SERVER}).set_env("LLAMA_ARG_TOOLS")); + add_opt(common_arg( + {"--mcp-servers-config"}, "PATH", + "experimental: path to JSON file with MCP server definitions (Cursor-compatible format) - do not enable in untrusted environments (default: none)\n" + "note: for security reasons, this will limit --cors-origins to localhost by default", + [](common_params & params, const std::string & value) { + params.mcp_servers_config = value; + } + ).set_examples({LLAMA_EXAMPLE_SERVER}).set_env("LLAMA_ARG_MCP_SERVERS_CONFIG")); + add_opt(common_arg( + {"--mcp-servers-json"}, "JSON", + "experimental: inline JSON with MCP server definitions (Cursor-compatible format) - do not enable in untrusted environments (default: none)\n" + "note: for security reasons, this will limit --cors-origins to localhost by default", + [](common_params & params, const std::string & value) { + params.mcp_servers_json = value; + } + ).set_examples({LLAMA_EXAMPLE_SERVER}).set_env("LLAMA_ARG_MCP_SERVERS_JSON")); add_opt(common_arg( {"-ag", "--agent"}, {"-no-ag", "--no-agent"}, diff --git a/common/common.h b/common/common.h index b5687c108..279252183 100644 --- a/common/common.h +++ b/common/common.h @@ -668,6 +668,10 @@ struct common_params { // enable built-in tools std::vector server_tools; + // MCP server configs (Cursor-compatible JSON) + std::string mcp_servers_config; // path to JSON file with MCP server definitions + std::string mcp_servers_json; // inline JSON with MCP server definitions + // router server configs std::string models_dir = ""; // directory containing models for the router server std::string models_preset = ""; // directory containing model presets for the router server diff --git a/tools/server/CMakeLists.txt b/tools/server/CMakeLists.txt index b5c40884f..280bd9e19 100644 --- a/tools/server/CMakeLists.txt +++ b/tools/server/CMakeLists.txt @@ -19,6 +19,8 @@ add_library(${TARGET} STATIC server-stream.h server-tools.cpp server-tools.h + server-mcp.cpp + server-mcp.h server-schema.cpp server-schema.h ) diff --git a/tools/server/README-dev.md b/tools/server/README-dev.md index e81336e5e..b4ec9f17d 100644 --- a/tools/server/README-dev.md +++ b/tools/server/README-dev.md @@ -189,7 +189,7 @@ This endpoint is intended to be used internally by the Web UI and subject to cha Get a list of tools, each tool has these fields: - `tool` (string): the ID name of the tool, to be used in POST call. Example: `read_file` - `display_name` (string): the name to be displayed on UI. Example: `Read file` -- `type` (string): always be `"builtin"` for now +- `type` (string): `"builtin"` for a built-in tool, or `"mcp"` for a tool exposed by an MCP server - `permissions` (object): a mapping string --> boolean that indicates the permission required by this tool. This is useful for the UI to ask the user before calling the tool. For now, the only permission supported is `"write"` - `definition` (object): the OAI-compat definition of this tool @@ -199,7 +199,7 @@ Invoke a tool call, request body is a JSON object with: - `tool` (string): the name of the tool - `params` (object): a mapping from argument name (string) to argument value -Returns JSON object. There are two response formats: +Returns JSON object. There are two response formats (MCP tools use the same two formats: their result content is concatenated into `plain_text_response`, and RPC or tool errors are surfaced as the `error` string): Format 1: Plain text. The text will be placed into a field called `plain_text_response`, example: diff --git a/tools/server/server-common.h b/tools/server/server-common.h index 583736638..6ef797ebb 100644 --- a/tools/server/server-common.h +++ b/tools/server/server-common.h @@ -9,9 +9,15 @@ #define JSON_ASSERT GGML_ASSERT #include +#include +#include +#include +#include +#include +#include +#include #include #include -#include using json = nlohmann::ordered_json; @@ -376,3 +382,67 @@ server_tokens format_prompt_rerank( mtmd_context * mctx, const std::string & query, const std::string & doc); + +// simple implementation of a pipe +// used for streaming data between threads +template +struct server_pipe { + std::mutex mutex; + std::condition_variable cv; + std::queue queue; + std::atomic writer_closed{false}; + std::atomic reader_closed{false}; + + // 0 = unbounded (default) + // > 0, write() drops the oldest item once the queue is full + size_t max_size = 0; + + void close_write() { + writer_closed.store(true, std::memory_order_relaxed); + cv.notify_all(); + } + + void close_read() { + reader_closed.store(true, std::memory_order_relaxed); + cv.notify_all(); + } + + // close_on_stop = true: should_stop means the reader is gone for good, so the writer is told the pipe is broken. + // close_on_stop = false: should_stop is a per-read deadline and further reads still come, so the pipe stays usable. + bool read(T & output, const std::function & should_stop, bool close_on_stop = true) { + std::unique_lock lk(mutex); + constexpr auto poll_interval = std::chrono::milliseconds(500); + while (true) { + if (!queue.empty()) { + output = std::move(queue.front()); + queue.pop(); + return true; + } + if (writer_closed.load()) { + return false; // clean EOF + } + if (should_stop && should_stop()) { // a null should_stop means "never stop" + if (close_on_stop) { + close_read(); // signal broken pipe to writer + } + return false; // cancelled / deadline reached + } + cv.wait_for(lk, poll_interval); + } + } + + bool write(T && data) { + std::lock_guard lk(mutex); + if (reader_closed.load()) { + return false; // broken pipe + } + if (max_size > 0) { + while (queue.size() >= max_size) { + queue.pop(); // drop oldest to stay bounded + } + } + queue.push(std::move(data)); + cv.notify_one(); + return true; + } +}; diff --git a/tools/server/server-mcp.cpp b/tools/server/server-mcp.cpp new file mode 100644 index 000000000..aa87afeb0 --- /dev/null +++ b/tools/server/server-mcp.cpp @@ -0,0 +1,836 @@ +#include "server-mcp.h" + +#include + +#include +#include +#include +#include +#include +#include +#include + +#if defined(_WIN32) +# include +# include +#else +# include +# include +# include +# include +extern char ** environ; +#endif + +// read NDJSON lines from a child pipe, calling on_line per line until `running` clears, EOF/error, or on_line returns false. +// polled, not blocking: a grandchild can inherit the pipe's write end and hold it open (terminate() kills only the direct child), so a blocking read would hang teardown on an EOF that never comes. +static void mcp_pump_ndjson(FILE * f, std::atomic & running, + const std::function & on_line) { + if (!f) { + return; + } + const int poll_ms = 50; + const size_t max_line = 8 * 1024 * 1024; // drop any single NDJSON line larger than this, so a child that never emits '\n' can't grow buf without bound +#if defined(_WIN32) + HANDLE h = (HANDLE) _get_osfhandle(_fileno(f)); +#else + int fd = fileno(f); + int fl = fcntl(fd, F_GETFL, 0); + if (fl >= 0) { + fcntl(fd, F_SETFL, fl | O_NONBLOCK); + } +#endif + std::string buf; + bool skipping = false; // discarding an over-long line until its terminating newline + char chunk[4096]; + while (running.load()) { + size_t n = 0; +#if defined(_WIN32) + DWORD avail = 0; + if (!PeekNamedPipe(h, NULL, 0, NULL, &avail, NULL)) { + break; // pipe broken / child gone + } + if (avail == 0) { + std::this_thread::sleep_for(std::chrono::milliseconds(poll_ms)); + continue; + } + DWORD to_read = avail < (DWORD) sizeof(chunk) ? avail : (DWORD) sizeof(chunk); + DWORD got = 0; + if (!ReadFile(h, chunk, to_read, &got, NULL) || got == 0) { + break; + } + n = (size_t) got; +#else + struct pollfd pfd; + pfd.fd = fd; + pfd.events = POLLIN; + pfd.revents = 0; + int pr = poll(&pfd, 1, poll_ms); + if (pr < 0) { + if (errno == EINTR) { + continue; + } + break; + } + if (pr == 0) { + continue; // timeout -> re-check running + } + if (pfd.revents & (POLLERR | POLLNVAL)) { + break; + } + ssize_t r = read(fd, chunk, sizeof(chunk)); + if (r < 0) { + if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) { + continue; + } + break; + } + if (r == 0) { + break; // EOF: child (and any pipe writers) closed the stream + } + n = (size_t) r; +#endif + buf.append(chunk, n); + + // resync after an over-long, unterminated line: discard bytes until the next newline + if (skipping) { + size_t nl = buf.find('\n'); + if (nl == std::string::npos) { + if (buf.size() > max_line) { + buf.clear(); // stay bounded while waiting for a terminator + } + continue; + } + buf.erase(0, nl + 1); + skipping = false; + } + + size_t pos; + while ((pos = buf.find('\n')) != std::string::npos) { + std::string line = buf.substr(0, pos); + buf.erase(0, pos + 1); + if (!line.empty() && line.back() == '\r') { + line.pop_back(); + } + if (line.empty()) { + continue; + } + if (!on_line(std::move(line))) { + return; + } + } + + // a partial line already larger than the cap and still no newline: drop it to avoid unbounded growth + if (buf.size() > max_line) { + SRV_WRN("MCP: dropping oversized line (> %zu bytes) from child pipe\n", max_line); + buf.clear(); + skipping = true; + } + } +} + +// +// server_mcp_server_config +// + +std::vector server_mcp_server_config::parse_from_json(const std::string & json_str) { + return parse_cursor_format(json::parse(json_str)); +} + +std::vector server_mcp_server_config::parse_cursor_format(const json & j) { + std::vector result; + + if (!j.contains("mcpServers") || !j.at("mcpServers").is_object()) { + return result; + } + + for (const auto & [name, cfg] : j.at("mcpServers").items()) { + server_mcp_server_config sc; + sc.name = name; + sc.command = cfg.value("command", std::string()); + sc.cwd = cfg.value("cwd", std::string()); + sc.timeout_ms = cfg.value("timeout_ms", sc.timeout_ms); + + if (cfg.contains("args") && cfg.at("args").is_array()) { + for (const auto & a : cfg.at("args")) { + sc.args.push_back(a.get()); + } + } + if (cfg.contains("env") && cfg.at("env").is_object()) { + for (const auto & [k, v] : cfg.at("env").items()) { + sc.env[k] = v.get(); + } + } + + if (sc.command.empty()) { + SRV_WRN("MCP server '%s' has no command, skipping\n", name.c_str()); + continue; + } + result.push_back(std::move(sc)); + } + + return result; +} + + +// +// server_mcp_transport +// + +static constexpr const char * MCP_PROTOCOL_VERSION = "2024-11-05"; + +static std::string rpc_error_message(const json & resp) { + if (resp.contains("error")) { + const json & e = resp.at("error"); + if (e.is_object()) { + return e.value("message", "unknown error"); + } + if (e.is_string()) { + return e.get(); + } + } + return "unknown error"; +} + +// normalize an MCP tools/call result to the /tools contract (see README-dev.md): +// concat text parts of result.content[], and surface an isError result +static json mcp_result_to_response(const json & result) { + std::string text; + if (result.contains("content") && result.at("content").is_array()) { + for (const auto & part : result.at("content")) { + if (part.is_object() && part.value("type", "") == "text") { + if (!text.empty()) { + text += "\n"; + } + text += part.value("text", ""); + } + } + } + if (result.is_object() && result.value("isError", false)) { + return {{"error", text.empty() ? "MCP tool returned an error" : text}}; + } + return {{"plain_text_response", text}}; +} + +json server_mcp_transport::send_rpc(const json & request, const std::function & should_stop) { + if (!to_server.write(request.dump())) { + return {{"error", {{"code", -32603}, {"message", "transport closed"}}}}; + } + + const bool has_id = request.contains("id"); + const auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(timeout_ms); + auto stop = [&]() { + return (should_stop && should_stop()) || std::chrono::steady_clock::now() >= deadline; + }; + + std::string frame; + while (from_server.read(frame, stop, false)) { + json reply; + try { + reply = json::parse(frame); + } catch (...) { + if (std::chrono::steady_clock::now() >= deadline) { + break; + } + continue; // skip malformed frame + } + // no id: a notification. mismatched id: a stale reply from a timed-out request (ids are monotonic, never a future one) + if (!has_id || (reply.contains("id") && reply.at("id") == request.at("id"))) { + return reply; + } + if (std::chrono::steady_clock::now() >= deadline) { + break; // a flood of notifications must not outrun the deadline + } + } + + if (should_stop && should_stop()) { + return {{"error", {{"code", -32603}, {"message", "cancelled"}}}}; + } + if (std::chrono::steady_clock::now() >= deadline) { + return {{"error", {{"code", -32603}, {"message", "request timed out"}}}}; + } + return {{"error", {{"code", -32603}, {"message", "transport closed"}}}}; +} + +bool server_mcp_transport::ensure_init(const std::function & should_stop) { + if (initialized) { + return true; + } + + json init_req = { + {"jsonrpc", "2.0"}, + {"id", next_id++}, + {"method", "initialize"}, + {"params", { + {"protocolVersion", MCP_PROTOCOL_VERSION}, + {"capabilities", json::object()}, + {"clientInfo", {{"name", "llama.cpp"}, {"version", "1.0"}}}, + }}, + }; + json resp = send_rpc(init_req, should_stop); + if (!resp.contains("result")) { + last_error = "initialize failed: " + rpc_error_message(resp); + return false; + } + + // notifications/initialized: no id, no reply expected + json notif = {{"jsonrpc", "2.0"}, {"method", "notifications/initialized"}}; + to_server.write(notif.dump()); + + initialized = true; + return true; +} + +std::vector server_mcp_transport::list_tools(const std::function & should_stop) { + std::lock_guard lock(rpc_mutex); + if (!ensure_init(should_stop)) { + return {}; + } + if (!tools.empty()) { + return tools; + } + + json req = {{"jsonrpc", "2.0"}, {"id", next_id++}, {"method", "tools/list"}}; + json resp = send_rpc(req, should_stop); + if (!resp.contains("result")) { + last_error = "tools/list failed: " + rpc_error_message(resp); + return {}; + } + + const json & result = resp.at("result"); + if (result.contains("tools") && result.at("tools").is_array()) { + for (const auto & t : result.at("tools")) { + server_mcp_tool_def def; + def.server_name = name; + def.name = t.value("name", ""); + def.description = t.value("description", ""); + if (t.contains("inputSchema")) { + def.input_schema = t.at("inputSchema"); + } + tools.push_back(std::move(def)); + } + } + return tools; +} + +json server_mcp_transport::call_tool(const std::string & tool_name, + const json & arguments, + const std::function & should_stop) { + std::lock_guard lock(rpc_mutex); + if (!ensure_init(should_stop)) { + return {{"error", last_error}}; + } + + json req = { + {"jsonrpc", "2.0"}, + {"id", next_id++}, + {"method", "tools/call"}, + {"params", {{"name", tool_name}, {"arguments", arguments}}}, + }; + json resp = send_rpc(req, should_stop); + if (resp.contains("error")) { + return {{"error", rpc_error_message(resp)}}; + } + if (resp.contains("result")) { + return mcp_result_to_response(resp.at("result")); + } + return {{"error", "invalid response from MCP server"}}; +} + +// +// server_mcp_stdio +// + +struct server_mcp_stdio::process_handle { + subprocess_s sp; + FILE * in = nullptr; // child stdin + FILE * out = nullptr; // child stdout + FILE * err = nullptr; // child stderr +}; + +#if defined(_WIN32) +// config strings are UTF-8 (from JSON) and subprocess.h converts them with CP_UTF8, so inputs must be UTF-8, not the active code page +static std::wstring windows_utf8_to_wide(const std::string & s) { + if (s.empty()) { + return std::wstring(); + } + int n = MultiByteToWideChar(CP_UTF8, 0, s.data(), (int) s.size(), NULL, 0); + if (n <= 0) { + return std::wstring(); + } + std::wstring w((size_t) n, L'\0'); + MultiByteToWideChar(CP_UTF8, 0, s.data(), (int) s.size(), &w[0], n); + return w; +} + +static std::string windows_wide_to_utf8(const wchar_t * s, int len /* -1 for NUL-terminated */) { + int n = WideCharToMultiByte(CP_UTF8, 0, s, len, NULL, 0, NULL, NULL); + if (n <= 0) { + return std::string(); + } + std::string out((size_t) n, '\0'); + WideCharToMultiByte(CP_UTF8, 0, s, len, &out[0], n, NULL, NULL); + if (len == -1 && !out.empty() && out.back() == '\0') { + out.pop_back(); // drop the terminator WideCharToMultiByte counts for -1 + } + return out; +} +#endif + +static std::string mcp_resolve_command(const std::string & command) { +#if defined(_WIN32) + // For Windows: make sure we handle ".exe" correctly, as well as UTF-8 + std::wstring wcmd = windows_utf8_to_wide(command); + wchar_t buf[MAX_PATH * 4]; + const DWORD cap = (DWORD) (sizeof(buf) / sizeof(buf[0])); + + auto search = [&](const wchar_t * ext) -> std::string { + DWORD n = SearchPathW(NULL, wcmd.c_str(), ext, cap, buf, NULL); + return (n > 0 && n < cap) ? windows_wide_to_utf8(buf, (int) n) : std::string(); + }; + + std::string found = search(NULL); // exact path / already-extensioned / .exe on PATH + if (!found.empty()) { + return found; + } + + std::wstring pathext; + DWORD need = GetEnvironmentVariableW(L"PATHEXT", NULL, 0); + if (need > 0) { + pathext.resize(need); + DWORD got = GetEnvironmentVariableW(L"PATHEXT", &pathext[0], need); + pathext.resize(got); + } + if (pathext.empty()) { + pathext = L".COM;.EXE;.BAT;.CMD"; + } + for (size_t start = 0; start <= pathext.size();) { + size_t sep = pathext.find(L';', start); + std::wstring ext = pathext.substr(start, sep == std::wstring::npos ? std::wstring::npos : sep - start); + if (!ext.empty()) { + found = search(ext.c_str()); + if (!found.empty()) { + return found; + } + } + if (sep == std::wstring::npos) { + break; + } + start = sep + 1; + } + return command; // give up and let subprocess.h report the spawn error +#else + return command; +#endif // _WIN32 +} + +static std::vector mcp_parent_env() { + std::vector env; +#if defined(_WIN32) + LPWCH block = GetEnvironmentStringsW(); + if (block) { + for (LPWCH e = block; *e; e += wcslen(e) + 1) { + env.emplace_back(windows_wide_to_utf8(e, -1)); + } + FreeEnvironmentStringsW(block); + } +#else + if (environ) { + for (char ** e = environ; *e; ++e) { + env.emplace_back(*e); + } + } +#endif + return env; +} + +// parent env with the config overrides applied, in "KEY=VALUE" form +static std::vector mcp_build_env(const std::map & overrides) { + std::vector env; + for (auto & e : mcp_parent_env()) { + size_t eq = e.find('='); + std::string key = eq == std::string::npos ? e : e.substr(0, eq); + if (overrides.find(key) == overrides.end()) { + env.push_back(e); + } + } + for (auto & [k, v] : overrides) { + env.push_back(k + "=" + v); + } + return env; +} + +server_mcp_stdio::server_mcp_stdio(const server_mcp_server_config & config) : config(config) { + name = config.name; + timeout_ms = config.timeout_ms; + // bound the reply queue: send_rpc only drains during a call, so unsolicited notifications would otherwise grow it without limit + from_server.max_size = 65536; +} + +server_mcp_stdio::~server_mcp_stdio() { + join_pumps(); +} + +bool server_mcp_stdio::start() { + std::vector argv_s; + argv_s.push_back(mcp_resolve_command(config.command)); + argv_s.insert(argv_s.end(), config.args.begin(), config.args.end()); + + int options = subprocess_option_no_window | subprocess_option_search_user_path; + std::vector envp_s; + if (config.env.empty()) { + options |= subprocess_option_inherit_environment; + } else { + envp_s = mcp_build_env(config.env); + } + + auto to_ptrs = [](std::vector & v) { + std::vector p; + p.reserve(v.size() + 1); + for (auto & s : v) { + p.push_back(s.c_str()); + } + p.push_back(nullptr); + return p; + }; + auto argv = to_ptrs(argv_s); + auto envp = to_ptrs(envp_s); + + auto handle = std::make_unique(); + int rc = subprocess_create_ex(argv.data(), options, + config.env.empty() ? nullptr : envp.data(), + config.cwd.empty() ? nullptr : config.cwd.c_str(), + &handle->sp); + if (rc != 0) { + SRV_WRN("MCP '%s': failed to spawn '%s'\n", config.name.c_str(), config.command.c_str()); + return false; + } + handle->in = subprocess_stdin(&handle->sp); + handle->out = subprocess_stdout(&handle->sp); + handle->err = subprocess_stderr(&handle->sp); + + proc = std::move(handle); + running.store(true); + reader = std::thread([this] { reader_loop(); }); + writer = std::thread([this] { writer_loop(); }); + errlog = std::thread([this] { errlog_loop(); }); + return true; +} + +void server_mcp_stdio::close() { + join_pumps(); +} + +bool server_mcp_stdio::is_alive() const { + return running.load(); +} + +std::string server_mcp_stdio::diagnostics() { + std::string out; + { + std::lock_guard lock(rpc_mutex); // last_error is written by send_rpc's callers + out = last_error; + } + std::lock_guard lk(err_mu); + if (!err_tail.empty()) { + if (!out.empty()) { + out += "; "; + } + out += "last stderr: " + err_tail; + } + return out; +} + +void server_mcp_stdio::reader_loop() { + mcp_pump_ndjson(proc->out, running, [this](std::string && line) { + return from_server.write(std::move(line)); // false => consumer gone, stop + }); + running.store(false); + to_server.close_write(); // stop the writer + from_server.close_write(); // EOF to any waiting caller +} + +// write all of `data` to child stdin, non-blocking and polled so teardown never hangs (a grandchild can hold the read end of a full pipe open). returns false on error/close/shutdown. +static bool mcp_write_all(FILE * f, const std::string & data, std::atomic & running) { + if (!f) { + return false; + } + size_t total = 0; +#if defined(_WIN32) + HANDLE h = (HANDLE) _get_osfhandle(_fileno(f)); + DWORD nowait = PIPE_NOWAIT; + SetNamedPipeHandleState(h, &nowait, NULL, NULL); + while (total < data.size() && running.load()) { + DWORD written = 0; + BOOL ok = WriteFile(h, data.data() + total, (DWORD) (data.size() - total), &written, NULL); + if (ok && written > 0) { + total += written; + continue; + } + if (!ok) { + DWORD err = GetLastError(); + if (err != ERROR_NO_DATA && err != ERROR_PIPE_BUSY) { + return false; + } + } + // backpressure (pipe full) is rare for small JSON-RPC frames; sleep rather than spin. + // no writable-wait exists for a PIPE_NOWAIT anonymous pipe, so this polls like the POSIX poll() path. + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } +#else + int fd = fileno(f); + int fl = fcntl(fd, F_GETFL, 0); + if (fl >= 0) { + fcntl(fd, F_SETFL, fl | O_NONBLOCK); + } + while (total < data.size() && running.load()) { + ssize_t n = write(fd, data.data() + total, data.size() - total); + if (n > 0) { + total += (size_t) n; + continue; + } + if (n == 0) { + return false; + } + if (errno == EINTR) { + continue; + } + if (errno != EAGAIN && errno != EWOULDBLOCK) { + return false; + } + struct pollfd pfd; + pfd.fd = fd; + pfd.events = POLLOUT; + pfd.revents = 0; + int pr = poll(&pfd, 1, 50); + if (pr < 0) { + if (errno == EINTR) { + continue; + } + return false; + } + if (pfd.revents & (POLLERR | POLLNVAL | POLLHUP)) { + return false; + } + } +#endif + return total == data.size(); +} + +void server_mcp_stdio::writer_loop() { + auto should_stop = [this] { return !running.load(); }; + std::string msg; + while (to_server.read(msg, should_stop)) { + msg.push_back('\n'); + if (!mcp_write_all(proc->in, msg, running)) { + break; // child gone or shutting down + } + } + running.store(false); + to_server.close_read(); // fail fast on any further send_rpc write + from_server.close_write(); // wake any caller waiting for a reply +} + +void server_mcp_stdio::errlog_loop() { + static constexpr size_t ERR_TAIL_MAX = 4096; + // drain stderr (an undrained pipe blocks the child): + // log it, and keep a bounded tail for reporting when the server dies + mcp_pump_ndjson(proc->err, running, [this](std::string && line) { + SRV_DBG("MCP '%s' stderr: %s\n", name.c_str(), line.c_str()); + std::lock_guard lk(err_mu); + err_tail += line; + err_tail += '\n'; + if (err_tail.size() > ERR_TAIL_MAX) { + err_tail.erase(0, err_tail.size() - ERR_TAIL_MAX); + } + return true; + }); +} + +void server_mcp_stdio::join_pumps() { + if (!proc) { + return; + } + running.store(false); + to_server.close_write(); // wake the writer if it waits for a message + from_server.close_write(); // wake any caller waiting for a reply + + subprocess_terminate(&proc->sp); // child death unblocks the blocked fread/fwrite + + if (writer.joinable()) writer.join(); + if (reader.joinable()) reader.join(); + if (errlog.joinable()) errlog.join(); + + subprocess_join(&proc->sp, nullptr); // reap the child: destroy() never waits, so the pid would stay a zombie for the process lifetime + subprocess_destroy(&proc->sp); // safe now: no thread touches the FILE* anymore + proc.reset(); +} + + +// +// server_mcp +// + +static constexpr int MCP_COOLDOWN_SECONDS = 5; +static constexpr int MCP_WARMUP_TIMEOUT_SECONDS = 10; // cap per-server tool discovery at startup + +server_mcp::~server_mcp() { + shutdown(); + + std::vector> to_close; + { + std::lock_guard lock(mutex); + for (auto & [name, t] : transports) { + to_close.push_back(std::move(t)); + } + transports.clear(); + } + for (auto & t : to_close) { + t->close(); + } +} + +std::shared_ptr server_mcp::create_transport(const server_mcp_server_config & cfg) { + return std::make_shared(cfg); +} + +void server_mcp::shutdown() { + stopping.store(true); +} + +const server_mcp_server_config * server_mcp::find_config(const std::string & name) const { + for (const auto & c : configs) { + if (c.name == name) { + return &c; + } + } + return nullptr; +} + +void server_mcp::start(const common_params & params) { + auto append = [this](const std::string & json_str) { + try { + auto parsed = server_mcp_server_config::parse_from_json(json_str); + if (parsed.empty()) { + SRV_WRN("%s", "MCP config: no servers found in JSON\n"); + } + for (auto & p : parsed) { + // names must be unique across both config sources: get_or_create / find_config key on the name + if (find_config(p.name)) { + SRV_WRN("MCP config: duplicate server name '%s', skipping\n", p.name.c_str()); + continue; + } + configs.push_back(std::move(p)); + } + } catch (const std::exception & e) { + throw std::runtime_error(std::string("failed to parse MCP config JSON: ") + e.what()); + } + }; + if (!params.mcp_servers_config.empty()) { + std::ifstream f = fs_open_ifstream(params.mcp_servers_config, std::ios::in); + if (!f) { + throw std::runtime_error("failed to open MCP config file: " + params.mcp_servers_config); + } + std::stringstream ss; + ss << f.rdbuf(); + append(ss.str()); + } + if (!params.mcp_servers_json.empty()) { + append(params.mcp_servers_json); + } + + if (configs.empty()) { + return; + } + + std::vector discovered; + for (const auto & cfg : configs) { + auto t = create_transport(cfg); + if (!t->start()) { + SRV_WRN("MCP warmup: failed to spawn '%s': %s\n", cfg.name.c_str(), t->diagnostics().c_str()); + continue; + } + // bound warmup per server so an unresponsive one can't stall startup for the full per-call timeout + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(MCP_WARMUP_TIMEOUT_SECONDS); + auto should_stop = [this, deadline]() { + return stopping.load() || std::chrono::steady_clock::now() >= deadline; + }; + auto tools = t->list_tools(should_stop); + SRV_INF("MCP warmup: '%s' discovered %zu tools\n", cfg.name.c_str(), tools.size()); + discovered.insert(discovered.end(), tools.begin(), tools.end()); + t->close(); + } + + std::lock_guard lock(mutex); + registry.swap(discovered); +} + +std::vector server_mcp::list_tools() const { + std::lock_guard lock(mutex); + return registry; +} + +json server_mcp::call_tool(const std::string & server_name, + const std::string & tool_name, + const json & arguments, + const std::function & should_stop) { + auto transport = get_or_create(server_name); + if (!transport) { + return {{"error", "MCP server unavailable: " + server_name}}; + } + + auto stop = [this, &should_stop]() { + return stopping.load() || (should_stop && should_stop()); + }; + return transport->call_tool(tool_name, arguments, stop); +} + +std::shared_ptr server_mcp::get_or_create(const std::string & name) { + std::vector> to_close; // closed after unlock + std::shared_ptr result; + + { + std::lock_guard lock(mutex); + if (stopping.load()) { + return nullptr; + } + + auto now = std::chrono::steady_clock::now(); + auto dead_it = dead_servers.find(name); + if (dead_it != dead_servers.end()) { + if (now < dead_it->second) { + return nullptr; + } + dead_servers.erase(dead_it); + } + + auto it = transports.find(name); + if (it != transports.end()) { + if (it->second->is_alive()) { + return it->second; + } + SRV_WRN("MCP '%s' is no longer alive: %s\n", name.c_str(), it->second->diagnostics().c_str()); + to_close.push_back(std::move(it->second)); + transports.erase(it); + } + + const server_mcp_server_config * cfg = find_config(name); + if (cfg) { + auto fresh = create_transport(*cfg); + if (fresh->start() && fresh->is_alive()) { + transports[name] = fresh; + result = fresh; + } else { + SRV_WRN("MCP '%s': failed to start: %s\n", name.c_str(), fresh->diagnostics().c_str()); + to_close.push_back(std::move(fresh)); + dead_servers[name] = now + std::chrono::seconds(MCP_COOLDOWN_SECONDS); + } + } + } + + for (auto & t : to_close) { + t->close(); // blocking call, no leaks + } + + return result; +} + diff --git a/tools/server/server-mcp.h b/tools/server/server-mcp.h new file mode 100644 index 000000000..c7f33a3f7 --- /dev/null +++ b/tools/server/server-mcp.h @@ -0,0 +1,176 @@ +#pragma once + +#include "server-common.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +// +// Configuration (Cursor-compatible "mcpServers" JSON) +// + +struct server_mcp_server_config { + std::string name; // config key, e.g. "filesystem" + std::string command; + std::vector args; + std::map env; // merged over the parent env + std::string cwd; + int timeout_ms = 30000; // per-tool-call timeout + + // throw on parse errors; missing "mcpServers" yields an empty list; entries without a "command" are skipped + static std::vector parse_from_json(const std::string & json_str); + static std::vector parse_cursor_format(const json & j); +}; + +// a tool advertised by an MCP server +struct server_mcp_tool_def { + std::string server_name; + std::string name; // bare tool name, no "_" prefix + std::string description; + json input_schema; // JSON Schema for the arguments, or null +}; + +// +// server_mcp_transport: one MCP server session. +// +// caller --send_rpc--> to_server --[writer]--> framing --> server +// caller <--send_rpc-- from_server <--[reader]-- framing <-- server +// +// each queue item is one complete serialized JSON message. +// subclass owns byte I/O and framing; base owns JSON and the JSON-RPC session (handshake, id correlation). +// + +struct server_mcp_transport { + std::string name; + int timeout_ms = 30000; + + server_pipe to_server; // serialized messages we send to the server + server_pipe from_server; // serialized messages read from the server + + virtual ~server_mcp_transport() = default; + + virtual bool start() = 0; + virtual void close() = 0; // blocking and idempotent + virtual bool is_alive() const = 0; // never blocks behind an in-flight send_rpc() + + // human-readable diagnostics for logging when the transport fails/dies + // (example: last RPC error, plus any transport-specific detail) + // may run on a different thread than send_rpc(), so last_error is read under rpc_mutex + virtual std::string diagnostics() { + std::lock_guard lock(rpc_mutex); + return last_error; + } + + std::vector list_tools(const std::function & should_stop); + + json call_tool(const std::string & tool_name, + const json & arguments, + const std::function & should_stop); + +protected: + // per-transport: send_rpc() holds it across the reply wait, so sharing it would stall every server behind one slow call. guards all members below. + std::mutex rpc_mutex; + uint64_t next_id = 1; // reset to 1 per (re)spawn + bool initialized = false; + std::string last_error; + std::vector tools; + + // both assume rpc_mutex is already held by the public caller + bool ensure_init(const std::function & should_stop); // initialize handshake, once + json send_rpc(const json & request, const std::function & should_stop); // returns the reply or an {"error": ...} +}; + +// +// server_mcp_stdio: child process, NDJSON JSON-RPC over stdio (stderr drained to the debug log) +// + +struct server_mcp_stdio : server_mcp_transport { + explicit server_mcp_stdio(const server_mcp_server_config & config); + ~server_mcp_stdio() override; + + bool start() override; + void close() override; + bool is_alive() const override; + std::string diagnostics() override; + +private: + server_mcp_server_config config; + + // defined in the .cpp so stays out of this header + struct process_handle; + std::unique_ptr proc; + + std::thread reader; // child stdout -> NDJSON de-framing -> from_server + std::thread writer; // to_server -> NDJSON framing -> child stdin + std::thread errlog; // child stderr -> debug log (must be drained or the child blocks) + + // cleared by close() or by the reader on stdout EOF; read without rpc_mutex + std::atomic running{false}; + + // bounded tail of the child's stderr, for diagnostics when it dies + std::mutex err_mu; + std::string err_tail; + + void reader_loop(); + void writer_loop(); + void errlog_loop(); + void join_pumps(); +}; + +// +// server_mcp +// declare before the HTTP context so it outlives every /tools handler. +// + +class server_mcp { +public: + server_mcp() = default; + ~server_mcp(); + + // parse the MCP config from params (file and/or inline JSON), + // then spawn each server once, list its tools, and shut it down + // throws on config parse errors; spawn failures are logged. + void start(const common_params & params); + + // true until start() has parsed at least one server from the config + bool empty() const { return configs.empty(); } + + std::vector list_tools() const; + + // lazily (re)spawns the transport. returns the MCP result or an {"error": ...}. should_stop is OR-ed with the manager's cancel flag. + json call_tool(const std::string & server_name, + const std::string & tool_name, + const json & arguments, + const std::function & should_stop = nullptr); + + // flip the cancel flag so in-flight calls return; blocking teardown is in the destructor. call before the HTTP server drains. + // note: multiple calls are idempotent + void shutdown(); + +private: + std::vector configs; + + mutable std::mutex mutex; // guards transports, dead_servers, registry + + // shared_ptr: call_tool() hands a transport to the caller and drops the lock for the blocking RPC, so a concurrent evict/respawn must not destroy it mid-call + std::map> transports; + std::map dead_servers; // spawn-failure cooldown + std::vector registry; + + std::atomic stopping{false}; + + const server_mcp_server_config * find_config(const std::string & name) const; + + // the only place that names a concrete transport + std::shared_ptr create_transport(const server_mcp_server_config & cfg); + + // nullptr during cooldown or shutdown + std::shared_ptr get_or_create(const std::string & name); +}; diff --git a/tools/server/server-models.cpp b/tools/server/server-models.cpp index 9eac58e9d..ba6378814 100644 --- a/tools/server/server-models.cpp +++ b/tools/server/server-models.cpp @@ -1944,53 +1944,6 @@ void server_models_routes::init_routes() { // server_http_proxy // -// simple implementation of a pipe -// used for streaming data between threads -template -struct pipe_t { - std::mutex mutex; - std::condition_variable cv; - std::queue queue; - std::atomic writer_closed{false}; - std::atomic reader_closed{false}; - void close_write() { - writer_closed.store(true, std::memory_order_relaxed); - cv.notify_all(); - } - void close_read() { - reader_closed.store(true, std::memory_order_relaxed); - cv.notify_all(); - } - bool read(T & output, const std::function & should_stop) { - std::unique_lock lk(mutex); - constexpr auto poll_interval = std::chrono::milliseconds(500); - while (true) { - if (!queue.empty()) { - output = std::move(queue.front()); - queue.pop(); - return true; - } - if (writer_closed.load()) { - return false; // clean EOF - } - if (should_stop()) { - close_read(); // signal broken pipe to writer - return false; // cancelled / reader no longer alive - } - cv.wait_for(lk, poll_interval); - } - } - bool write(T && data) { - std::lock_guard lk(mutex); - if (reader_closed.load()) { - return false; // broken pipe - } - queue.push(std::move(data)); - cv.notify_one(); - return true; - } -}; - static std::string to_lower_copy(const std::string & value) { std::string lowered(value.size(), '\0'); std::transform(value.begin(), value.end(), lowered.begin(), [](unsigned char c) { return std::tolower(c); }); @@ -2100,7 +2053,7 @@ server_http_proxy::server_http_proxy( ) { // shared between reader and writer threads auto cli = std::make_shared(host, port); - auto pipe = std::make_shared>(); + auto pipe = std::make_shared>(); if (scheme == "https") { #ifdef CPPHTTPLIB_OPENSSL_SUPPORT diff --git a/tools/server/server-tools.cpp b/tools/server/server-tools.cpp index a82a3d602..2af44e49b 100644 --- a/tools/server/server-tools.cpp +++ b/tools/server/server-tools.cpp @@ -10,10 +10,10 @@ #include #include #include -#include #include #include #include +#include namespace fs = std::filesystem; @@ -25,7 +25,7 @@ json server_tool::to_json() const { return { {"display_name", display_name}, {"tool", name}, - {"type", "builtin"}, + {"type", type()}, {"permissions", json{ {"write", permission_write} }}, @@ -1129,6 +1129,49 @@ struct server_tools_res : server_http_res { } }; +// +// server_mcp_tool: exposes one tool from a running MCP server as a server_tool. +// +struct server_mcp_tool : server_tool { + std::string server_name; + std::string tool_name; + server_mcp_tool_def def; + server_mcp & mcp_mgr; + + server_mcp_tool(server_mcp_tool_def d, server_mcp & mgr) + : server_name(d.server_name) + , tool_name(d.name) + , def(std::move(d)) + , mcp_mgr(mgr) + { + name = server_name + "_" + tool_name; + display_name = name; + permission_write = false; + support_stream = false; + } + + std::string type() const override { return "mcp"; } + + json get_definition() const override { + json schema = def.input_schema; + if (schema.is_null() || !schema.is_object()) { + schema = json::object(); + } + return { + {"type", "function"}, + {"function", { + {"name", name}, + {"description", def.description}, + {"parameters", schema}, + }}, + }; + } + + json invoke(json params, server_tool::stream *) const override { + return mcp_mgr.call_tool(server_name, tool_name, params); + } +}; + static server_tool & find_tool(std::vector> & tools, const std::string & name, bool require_stream) { for (auto & t : tools) { if (t->name == name) { @@ -1157,7 +1200,8 @@ static std::vector> build_tools() { return tools; } -void server_tools::setup(const std::vector & enabled_tools) { +void server_tools::setup(const std::vector & enabled_tools, + server_mcp & mcp_mgr) { if (!enabled_tools.empty()) { std::unordered_set enabled_set(enabled_tools.begin(), enabled_tools.end()); auto all_tools = build_tools(); @@ -1188,6 +1232,29 @@ void server_tools::setup(const std::vector & enabled_tools) { } } + // append MCP tools, skipping any that collide with a built-in or another MCP tool of the same "_" name + if (!mcp_mgr.empty()) { + std::unordered_set seen_names; + for (auto & t : tools) { + seen_names.insert(t->name); + } + size_t n_added = 0; + for (const auto & def : mcp_mgr.list_tools()) { + std::string mcp_name = def.server_name + "_" + def.name; + if (seen_names.count(mcp_name)) { + SRV_WRN("MCP tool \"%s\" from server \"%s\" collides with an existing tool, skipping\n", + mcp_name.c_str(), def.server_name.c_str()); + continue; + } + seen_names.insert(mcp_name); + tools.push_back(std::make_unique(def, mcp_mgr)); + n_added++; + } + if (n_added > 0) { + SRV_INF("Added %zu MCP tools\n", n_added); + } + } + handle_get = [this](const server_http_req &) -> server_http_res_ptr { auto res = std::make_unique(); try { diff --git a/tools/server/server-tools.h b/tools/server/server-tools.h index 6f6528f48..601399ee9 100644 --- a/tools/server/server-tools.h +++ b/tools/server/server-tools.h @@ -3,9 +3,11 @@ #include "server-common.h" #include "server-http.h" #include "server-queue.h" +#include "server-mcp.h" #include #include +#include struct server_tool { std::string name; @@ -15,6 +17,7 @@ struct server_tool { virtual ~server_tool() = default; virtual json get_definition() const = 0; + virtual std::string type() const { return "builtin"; } struct stream { server_response & qr; @@ -34,7 +37,8 @@ struct server_tools { server_response queue_res; std::atomic res_id{0}; - void setup(const std::vector & enabled_tools); + void setup(const std::vector & enabled_tools, + server_mcp & mcp_mgr); server_http_context::handler_t handle_get; server_http_context::handler_t handle_post; diff --git a/tools/server/server.cpp b/tools/server/server.cpp index 20effbb14..b6fef99e8 100644 --- a/tools/server/server.cpp +++ b/tools/server/server.cpp @@ -88,6 +88,11 @@ static server_http_context::handler_t ex_wrapper(server_http_context::handler_t int llama_server(int argc, char ** argv) { std::setlocale(LC_NUMERIC, "C"); +#ifndef _WIN32 + // Ignore SIGPIPE so the server does not crash if an MCP child exits while we are writing to its stdin + signal(SIGPIPE, SIG_IGN); +#endif + // own arguments required by this example common_params params; @@ -157,6 +162,9 @@ int llama_server(common_params & params, int argc, char ** argv) { params.model_alias.insert(model_name); } + // note: this is guaranteed to out-live ctx_http and tools + server_mcp mcp_mgr; + // struct that contains llama context and inference server_context ctx_server; @@ -326,17 +334,28 @@ int llama_server(common_params & params, int argc, char ** argv) { ctx_http.post("/cors-proxy", ex_wrapper(res_403)); } - // EXPERIMENTAL built-in tools - if (!params.server_tools.empty()) { + try { + mcp_mgr.start(params); + } catch (const std::exception & e) { + SRV_ERR("MCP starting failed: %s\n", e.what()); + return 1; + } + + if (!params.server_tools.empty() || !mcp_mgr.empty()) { try { - tools.setup(params.server_tools); + tools.setup(params.server_tools, mcp_mgr); } catch (const std::exception & e) { SRV_ERR("tools setup failed: %s\n", e.what()); return 1; } ctx_http.get ("/tools", ex_wrapper(tools.handle_get)); ctx_http.post("/tools", ex_wrapper(tools.handle_post)); - warn_names.push_back("built-in tools (experimental)"); + if (!params.server_tools.empty()) { + warn_names.push_back("built-in tools (experimental)"); + } + if (!mcp_mgr.empty()) { + warn_names.push_back("MCP servers (experimental)"); + } } else { ctx_http.get ("/tools", ex_wrapper(res_403)); ctx_http.post("/tools", ex_wrapper(res_403)); @@ -378,7 +397,7 @@ int llama_server(common_params & params, int argc, char ** argv) { if (is_router_server) { SRV_INF("%s", "starting server in router mode. models will be automatically loaded on-demand\n"); - clean_up = [&models_routes]() { + clean_up = [&models_routes, &mcp_mgr]() { SRV_INF("%s: cleaning up before exit...\n", __func__); // stop the session GC first, it finalizes live sessions and wakes pending readers server_stream_session_manager_stop(); @@ -386,6 +405,7 @@ int llama_server(common_params & params, int argc, char ** argv) { models_routes->stopping.store(true); // maybe redundant, but just to be safe models_routes->models.unload_all(); } + mcp_mgr.shutdown(); llama_backend_free(); }; @@ -401,17 +421,19 @@ int llama_server(common_params & params, int argc, char ** argv) { // important to disconnect any SSE clients models_routes->stopping.store(true); } + mcp_mgr.shutdown(); ctx_http.stop(); }; } else { // setup clean up function, to be called before exit - clean_up = [&ctx_http, &ctx_server]() { + clean_up = [&ctx_http, &ctx_server, &mcp_mgr]() { SRV_INF("%s: cleaning up before exit...\n", __func__); // stop the session GC first, it finalizes live sessions and wakes pending readers server_stream_session_manager_stop(); ctx_http.stop(); ctx_server.terminate(); + mcp_mgr.shutdown(); llama_backend_free(); }; @@ -444,6 +466,7 @@ int llama_server(common_params & params, int argc, char ** argv) { SRV_INF("%s", "model loaded\n"); shutdown_handler = [&](int) { + mcp_mgr.shutdown(); // this will unblock start_loop() ctx_server.terminate(); }; diff --git a/tools/server/tests/fixtures/mcp_burst_server.py b/tools/server/tests/fixtures/mcp_burst_server.py new file mode 100644 index 000000000..22892d9a1 --- /dev/null +++ b/tools/server/tests/fixtures/mcp_burst_server.py @@ -0,0 +1,118 @@ +#!/usr/bin/env python3 +""" +Minimal MCP server that writes notification + response in a single write() with no flush. +This reproduces the buffering bug where read_message() can strand the response. +""" +import json +import sys +import os + +TOOLS = [ + { + "name": "echo", + "description": "Echo back the input message", + "inputSchema": { + "type": "object", + "properties": { + "message": {"type": "string"} + }, + "required": ["message"] + } + } +] + +def handle_initialize(params, req_id): + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "protocolVersion": "2024-11-05", + "capabilities": {"tools": {}}, + "serverInfo": {"name": "burst-test", "version": "1.0"} + } + } + +def handle_tools_list(params, req_id): + return { + "jsonrpc": "2.0", + "id": req_id, + "result": {"tools": TOOLS} + } + +def handle_tools_call(params, req_id): + tool_name = params.get("name") + arguments = params.get("arguments", {}) + + if tool_name == "echo": + message = arguments.get("message", "") + notif = { + "jsonrpc": "2.0", + "method": "notifications/progress", + "params": {"progress": 50, "total": 100} + } + response = { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [{"type": "text", "text": f"echo: {message}"}] + } + } + # Single os.write() call: both lines land in one pipe packet atomically. + # This is the key difference from mcp_malformed_server.py which flushes between writes. + data = (json.dumps(notif) + "\n" + json.dumps(response) + "\n").encode("utf-8") + os.write(sys.stdout.fileno(), data) + return None # already written + else: + response = { + "jsonrpc": "2.0", + "id": req_id, + "error": {"code": -32602, "message": f"Unknown tool: {tool_name}"} + } + return response + +HANDLERS = { + "initialize": handle_initialize, + "tools/list": handle_tools_list, + "tools/call": handle_tools_call, +} + +def main(): + # Use line-buffered text mode for regular responses, but the burst write + # uses os.write() directly to guarantee a single kernel write(). + sys.stdout = os.fdopen(sys.stdout.fileno(), "w", buffering=1) + sys.stderr = os.fdopen(sys.stderr.fileno(), "w", buffering=1) + + for line in sys.stdin: + line = line.strip() + if not line: + continue + try: + request = json.loads(line) + except json.JSONDecodeError: + continue + + method = request.get("method") + req_id = request.get("id") + params = request.get("params", {}) + + # JSON-RPC 2.0: a message without an id is a notification and must not receive a response + if req_id is None: + continue + + handler = HANDLERS.get(method) + if handler: + response = handler(params, req_id) + if response is not None: + sys.stdout.write(json.dumps(response) + "\n") + sys.stdout.flush() + else: + response = { + "jsonrpc": "2.0", + "id": req_id, + "error": {"code": -32601, "message": f"Method not found: {method}"} + } + sys.stdout.write(json.dumps(response) + "\n") + sys.stdout.flush() + +if __name__ == "__main__": + main() diff --git a/tools/server/tests/fixtures/mcp_crash_server.py b/tools/server/tests/fixtures/mcp_crash_server.py new file mode 100644 index 000000000..8dffdc61c --- /dev/null +++ b/tools/server/tests/fixtures/mcp_crash_server.py @@ -0,0 +1,114 @@ +#!/usr/bin/env python3 +""" +MCP server that crashes after receiving a specific tool call. +""" +import json +import sys +import os + +def handle_initialize(params, req_id): + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "protocolVersion": "2024-11-05", + "capabilities": {"tools": {}}, + "serverInfo": {"name": "crash-test", "version": "1.0"} + } + } + +def handle_tools_list(params, req_id): + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "tools": [ + { + "name": "echo", + "description": "Echo back the input message", + "inputSchema": { + "type": "object", + "properties": { + "message": {"type": "string"} + } + } + }, + { + "name": "crash", + "description": "Crash the server", + "inputSchema": { + "type": "object", + "properties": {} + } + } + ] + } + } + +def handle_tools_call(params, req_id): + tool_name = params.get("name") + arguments = params.get("arguments", {}) + + if tool_name == "echo": + message = arguments.get("message", "") + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [{"type": "text", "text": f"echo: {message}"}] + } + } + elif tool_name == "crash": + # Send a partial response then exit + sys.stdout.write(json.dumps({"jsonrpc": "2.0", "id": req_id, "result": {"content": [{"type": "text", "text": "crashing..."}]}}) + "\n") + sys.stdout.flush() + os._exit(1) + else: + return { + "jsonrpc": "2.0", + "id": req_id, + "error": {"code": -32602, "message": f"Unknown tool: {tool_name}"} + } + +HANDLERS = { + "initialize": handle_initialize, + "tools/list": handle_tools_list, + "tools/call": handle_tools_call, +} + +def main(): + sys.stdout = os.fdopen(sys.stdout.fileno(), "w", buffering=1) + sys.stderr = os.fdopen(sys.stderr.fileno(), "w", buffering=1) + + for line in sys.stdin: + line = line.strip() + if not line: + continue + try: + request = json.loads(line) + except json.JSONDecodeError: + continue + + method = request.get("method") + req_id = request.get("id") + params = request.get("params", {}) + + # JSON-RPC 2.0: a message without an id is a notification and must not receive a response + if req_id is None: + continue + + handler = HANDLERS.get(method) + if handler: + response = handler(params, req_id) + else: + response = { + "jsonrpc": "2.0", + "id": req_id, + "error": {"code": -32601, "message": f"Method not found: {method}"} + } + + sys.stdout.write(json.dumps(response) + "\n") + sys.stdout.flush() + +if __name__ == "__main__": + main() diff --git a/tools/server/tests/fixtures/mcp_echo_server.py b/tools/server/tests/fixtures/mcp_echo_server.py new file mode 100755 index 000000000..7acfb3588 --- /dev/null +++ b/tools/server/tests/fixtures/mcp_echo_server.py @@ -0,0 +1,164 @@ +#!/usr/bin/env python3 +""" +Minimal MCP server for testing. +Implements JSON-RPC 2.0 over stdio (line-delimited JSON). +""" +import json +import sys +import os + +# Ensure we use python3 from the current environment +if sys.platform == "win32": + # On Windows, we need to use the same python interpreter + pass + +TOOLS = [ + { + "name": "echo", + "description": "Echo back the input message", + "inputSchema": { + "type": "object", + "properties": { + "message": {"type": "string", "description": "Message to echo"} + }, + "required": ["message"] + } + }, + { + "name": "add", + "description": "Add two numbers", + "inputSchema": { + "type": "object", + "properties": { + "a": {"type": "number"}, + "b": {"type": "number"} + }, + "required": ["a", "b"] + } + }, + { + "name": "fail_once", + "description": "Fails on first call, succeeds on subsequent calls", + "inputSchema": { + "type": "object", + "properties": {} + } + } +] + +_state = {"fail_once_called": False} + +def handle_initialize(params, req_id): + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "protocolVersion": "2024-11-05", + "capabilities": {"tools": {}}, + "serverInfo": {"name": "echo-test", "version": "1.0"} + } + } + +def handle_tools_list(params, req_id): + return { + "jsonrpc": "2.0", + "id": req_id, + "result": {"tools": TOOLS} + } + +def handle_tools_call(params, req_id): + tool_name = params.get("name") + arguments = params.get("arguments", {}) + + if tool_name == "echo": + message = arguments.get("message", "") + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [{"type": "text", "text": f"echo: {message}"}] + } + } + elif tool_name == "add": + a = arguments.get("a", 0) + b = arguments.get("b", 0) + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [{"type": "text", "text": str(a + b)}] + } + } + elif tool_name == "fail_once": + if not _state["fail_once_called"]: + _state["fail_once_called"] = True + return { + "jsonrpc": "2.0", + "id": req_id, + "error": {"code": -32000, "message": "transient error"} + } + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [{"type": "text", "text": "ok"}] + } + } + else: + return { + "jsonrpc": "2.0", + "id": req_id, + "error": {"code": -32602, "message": f"Unknown tool: {tool_name}"} + } + +def handle_ping(params, req_id): + return { + "jsonrpc": "2.0", + "id": req_id, + "result": {} + } + +HANDLERS = { + "initialize": handle_initialize, + "tools/list": handle_tools_list, + "tools/call": handle_tools_call, + "ping": handle_ping, +} + +def main(): + # Use unbuffered output + sys.stdout = os.fdopen(sys.stdout.fileno(), "w", buffering=1) + sys.stderr = os.fdopen(sys.stderr.fileno(), "w", buffering=1) + + for line in sys.stdin: + line = line.strip() + if not line: + continue + try: + request = json.loads(line) + except json.JSONDecodeError: + continue + + method = request.get("method") + req_id = request.get("id") + params = request.get("params", {}) + + # JSON-RPC 2.0: a message without an id is a notification and must not receive a response + if req_id is None: + continue + + handler = HANDLERS.get(method) + if handler: + response = handler(params, req_id) + else: + response = { + "jsonrpc": "2.0", + "id": req_id, + "error": {"code": -32601, "message": f"Method not found: {method}"} + } + + sys.stdout.write(json.dumps(response) + "\n") + sys.stdout.flush() + +if __name__ == "__main__": + main() diff --git a/tools/server/tests/fixtures/mcp_grandchild_server.py b/tools/server/tests/fixtures/mcp_grandchild_server.py new file mode 100644 index 000000000..2604a77ee --- /dev/null +++ b/tools/server/tests/fixtures/mcp_grandchild_server.py @@ -0,0 +1,100 @@ +#!/usr/bin/env python3 +""" +MCP server (NDJSON JSON-RPC over stdio) that spawns a long-lived grandchild which inherits +this process's stdin/stdout/stderr and keeps them open. + +This reproduces the reader-teardown deadlock: killing the direct MCP child (SIGKILL, which is +all subprocess_terminate() does) does NOT close the stdout/stderr pipe write ends, because the +grandchild still holds them. A server that reads those pipes with a blocking read would then +wait forever for an EOF that never arrives, hanging teardown (both warmup shutdown at startup +and process shutdown). The polled, running-aware reader must exit regardless. +""" +import json +import os +import subprocess +import sys + +# Spawn a grandchild that inherits our std handles (fds 0/1/2 = the MCP pipes) and lives well +# past any teardown in the tests. We do NOT redirect its stdio, so it keeps the pipe write ends +# open even after this process is killed. +subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"]) + +TOOLS = [ + { + "name": "echo", + "description": "Echo back the input message", + "inputSchema": { + "type": "object", + "properties": {"message": {"type": "string", "description": "Message to echo"}}, + "required": ["message"], + }, + } +] + + +def handle_initialize(params, req_id): + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "protocolVersion": "2024-11-05", + "capabilities": {"tools": {}}, + "serverInfo": {"name": "grandchild-test", "version": "1.0"}, + }, + } + + +def handle_tools_list(params, req_id): + return {"jsonrpc": "2.0", "id": req_id, "result": {"tools": TOOLS}} + + +def handle_tools_call(params, req_id): + if params.get("name") == "echo": + message = params.get("arguments", {}).get("message", "") + return { + "jsonrpc": "2.0", + "id": req_id, + "result": {"content": [{"type": "text", "text": f"echo: {message}"}]}, + } + return {"jsonrpc": "2.0", "id": req_id, "error": {"code": -32602, "message": "Unknown tool"}} + + +HANDLERS = { + "initialize": handle_initialize, + "tools/list": handle_tools_list, + "tools/call": handle_tools_call, +} + + +def main(): + sys.stdout = os.fdopen(sys.stdout.fileno(), "w", buffering=1) + sys.stderr = os.fdopen(sys.stderr.fileno(), "w", buffering=1) + + for line in sys.stdin: + line = line.strip() + if not line: + continue + try: + request = json.loads(line) + except json.JSONDecodeError: + continue + + method = request.get("method") + req_id = request.get("id") + params = request.get("params", {}) + + if req_id is None: + continue # notification, no response + + handler = HANDLERS.get(method) + if handler: + response = handler(params, req_id) + else: + response = {"jsonrpc": "2.0", "id": req_id, "error": {"code": -32601, "message": f"Method not found: {method}"}} + + sys.stdout.write(json.dumps(response) + "\n") + sys.stdout.flush() + + +if __name__ == "__main__": + main() diff --git a/tools/server/tests/fixtures/mcp_malformed_server.py b/tools/server/tests/fixtures/mcp_malformed_server.py new file mode 100644 index 000000000..743333c5f --- /dev/null +++ b/tools/server/tests/fixtures/mcp_malformed_server.py @@ -0,0 +1,113 @@ +#!/usr/bin/env python3 +""" +MCP server that sends malformed responses and notifications during requests. +""" +import json +import sys +import os + +def handle_initialize(params, req_id): + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "protocolVersion": "2024-11-05", + "capabilities": {"tools": {}}, + "serverInfo": {"name": "malformed-test", "version": "1.0"} + } + } + +def handle_tools_list(params, req_id): + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "tools": [ + { + "name": "echo", + "description": "Echo back the input message", + "inputSchema": { + "type": "object", + "properties": { + "message": {"type": "string"} + } + } + } + ] + } + } + +def handle_tools_call(params, req_id): + tool_name = params.get("name") + arguments = params.get("arguments", {}) + + if tool_name == "echo": + message = arguments.get("message", "") + # Send a notification first (no id field) + notif = { + "jsonrpc": "2.0", + "method": "notifications/progress", + "params": {"progress": 50, "total": 100} + } + sys.stdout.write(json.dumps(notif) + "\n") + sys.stdout.flush() + # Then send the actual response + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [{"type": "text", "text": f"echo: {message}"}] + } + } + else: + return { + "jsonrpc": "2.0", + "id": req_id, + "error": {"code": -32602, "message": f"Unknown tool: {tool_name}"} + } + +HANDLERS = { + "initialize": handle_initialize, + "tools/list": handle_tools_list, + "tools/call": handle_tools_call, +} + +def main(): + sys.stdout = os.fdopen(sys.stdout.fileno(), "w", buffering=1) + sys.stderr = os.fdopen(sys.stderr.fileno(), "w", buffering=1) + + for line in sys.stdin: + line = line.strip() + if not line: + continue + try: + request = json.loads(line) + except json.JSONDecodeError: + # Send malformed JSON response + sys.stdout.write("THIS IS NOT JSON\n") + sys.stdout.flush() + continue + + method = request.get("method") + req_id = request.get("id") + params = request.get("params", {}) + + # JSON-RPC 2.0: a message without an id is a notification and must not receive a response + if req_id is None: + continue + + handler = HANDLERS.get(method) + if handler: + response = handler(params, req_id) + else: + response = { + "jsonrpc": "2.0", + "id": req_id, + "error": {"code": -32601, "message": f"Method not found: {method}"} + } + + sys.stdout.write(json.dumps(response) + "\n") + sys.stdout.flush() + +if __name__ == "__main__": + main() diff --git a/tools/server/tests/fixtures/mcp_slow_server.py b/tools/server/tests/fixtures/mcp_slow_server.py new file mode 100644 index 000000000..7f8e67835 --- /dev/null +++ b/tools/server/tests/fixtures/mcp_slow_server.py @@ -0,0 +1,132 @@ +#!/usr/bin/env python3 +""" +MCP server that sleeps before responding, for timeout testing. +""" +import json +import sys +import os +import time +import argparse + +TOOLS = [ + { + "name": "sleep", + "description": "Sleep for a given number of seconds", + "inputSchema": { + "type": "object", + "properties": { + "seconds": {"type": "number", "description": "Seconds to sleep"} + }, + "required": ["seconds"] + } + } +] + +def handle_initialize(params, req_id): + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "protocolVersion": "2024-11-05", + "capabilities": {"tools": {}}, + "serverInfo": {"name": "slow-test", "version": "1.0"} + } + } + +def handle_tools_list(params, req_id): + return { + "jsonrpc": "2.0", + "id": req_id, + "result": {"tools": TOOLS} + } + +def handle_tools_call(params, req_id): + tool_name = params.get("name") + arguments = params.get("arguments", {}) + + if tool_name == "sleep": + seconds = arguments.get("seconds", 1) + time.sleep(seconds) + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [{"type": "text", "text": f"slept {seconds}s"}] + } + } + else: + return { + "jsonrpc": "2.0", + "id": req_id, + "error": {"code": -32602, "message": f"Unknown tool: {tool_name}"} + } + +HANDLERS = { + "initialize": handle_initialize, + "tools/list": handle_tools_list, + "tools/call": handle_tools_call, +} + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--delay", type=float, default=5.0, help="Delay in seconds for sleep tool") + args = parser.parse_args() + + # Override the sleep duration + global handle_tools_call + def handle_tools_call(params, req_id): + tool_name = params.get("name") + arguments = params.get("arguments", {}) + + if tool_name == "sleep": + seconds = arguments.get("seconds", args.delay) + time.sleep(seconds) + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "content": [{"type": "text", "text": f"slept {seconds}s"}] + } + } + else: + return { + "jsonrpc": "2.0", + "id": req_id, + "error": {"code": -32602, "message": f"Unknown tool: {tool_name}"} + } + + sys.stdout = os.fdopen(sys.stdout.fileno(), "w", buffering=1) + sys.stderr = os.fdopen(sys.stderr.fileno(), "w", buffering=1) + + for line in sys.stdin: + line = line.strip() + if not line: + continue + try: + request = json.loads(line) + except json.JSONDecodeError: + continue + + method = request.get("method") + req_id = request.get("id") + params = request.get("params", {}) + + # JSON-RPC 2.0: a message without an id is a notification and must not receive a response + if req_id is None: + continue + + handler = HANDLERS.get(method) + if handler: + response = handler(params, req_id) + else: + response = { + "jsonrpc": "2.0", + "id": req_id, + "error": {"code": -32601, "message": f"Method not found: {method}"} + } + + sys.stdout.write(json.dumps(response) + "\n") + sys.stdout.flush() + +if __name__ == "__main__": + main() diff --git a/tools/server/tests/unit/test_mcp_servers.py b/tools/server/tests/unit/test_mcp_servers.py new file mode 100644 index 000000000..9ad2241bd --- /dev/null +++ b/tools/server/tests/unit/test_mcp_servers.py @@ -0,0 +1,718 @@ +#!/usr/bin/env python3 +""" +Tests for MCP server integration via the /tools endpoint. + +Invariants verified: +1. MCP tools appear in /tools listing when configured +2. MCP tools use _ naming +3. MCP tools can be invoked and return correct results +4. Misconfigured MCP servers do not crash the server +5. Multiple MCP servers can be configured simultaneously +6. Warmup populates the tool list at startup +""" +import json +import os +import sys +import tempfile +import time + +import pytest + +from utils import * + +# Path to the test MCP server fixture +FIXTURES_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "fixtures") +MCP_ECHO_SERVER = os.path.join(FIXTURES_DIR, "mcp_echo_server.py") + +server: ServerProcess + + +def _mcp_config_json(servers: dict) -> str: + """Create a JSON config string for --mcp-servers-json.""" + return json.dumps({"mcpServers": servers}) + + +def _start_server_with_mcp(mcp_json: str, **kwargs) -> ServerProcess: + """Helper to start a router server with MCP config.""" + srv = ServerPreset.router() + srv.server_tools = "all" + srv.no_ui = True + srv.server_port = 8085 # avoid conflict with load_all() which uses 8080 + srv.mcp_servers_json = mcp_json + for k, v in kwargs.items(): + setattr(srv, k, v) + srv.start() + return srv + + +def test_mcp_tools_listed_in_tools_endpoint(): + """MCP tools should appear in GET /tools with server:tool naming.""" + global server + mcp_json = _mcp_config_json({ + "echo": { + "command": sys.executable, + "args": [MCP_ECHO_SERVER], + } + }) + server = _start_server_with_mcp(mcp_json) + + try: + res = server.make_request("GET", "/tools") + assert res.status_code == 200, res.body + + tools = res.body + assert isinstance(tools, list), f"Expected list, got {type(tools)}" + + # Find MCP tools - name is in "tool" field or definition.function.name + def get_tool_name(t): + return t.get("tool", "") or t.get("definition", {}).get("function", {}).get("name", "") + + mcp_tools = [t for t in tools if get_tool_name(t).startswith("echo_")] + assert len(mcp_tools) >= 2, f"Expected at least 2 echo_ tools, got {len(mcp_tools)}: {mcp_tools}" + + tool_names = {get_tool_name(t) for t in mcp_tools} + assert "echo_echo" in tool_names + assert "echo_add" in tool_names + + # Verify tool structure + echo_tool = next(t for t in mcp_tools if get_tool_name(t) == "echo_echo") + assert "description" in echo_tool or "definition" in echo_tool + finally: + server.stop() + + +def test_mcp_tool_invocation(): + """MCP tools should be callable via POST /tools and return correct results.""" + global server + mcp_json = _mcp_config_json({ + "echo": { + "command": sys.executable, + "args": [MCP_ECHO_SERVER], + } + }) + server = _start_server_with_mcp(mcp_json) + + try: + # Call echo_echo + res = server.make_request("POST", "/tools", data={ + "tool": "echo_echo", + "params": {"message": "hello world"} + }) + assert res.status_code == 200, res.body + body = res.body + assert "error" not in body, body + # The result format depends on the tool implementation + # For MCP tools, it should contain the tool result + assert "plain_text_response" in body or "result" in body or "content" in body, body + + # Call echo_add + res = server.make_request("POST", "/tools", data={ + "tool": "echo_add", + "params": {"a": 3, "b": 5} + }) + assert res.status_code == 200, res.body + body = res.body + assert "error" not in body, body + finally: + server.stop() + + +def test_mcp_bad_command_does_not_crash(): + """A misconfigured MCP server should not crash the llama-server.""" + global server + mcp_json = _mcp_config_json({ + "nonexistent": { + "command": "this_executable_does_not_exist_12345", + "args": [], + } + }) + server = _start_server_with_mcp(mcp_json) + + try: + # Server should still be healthy + res = server.make_request("GET", "/health") + assert res.status_code == 200, res.body + + # Builtin tools should still work + res = server.make_request("GET", "/tools") + assert res.status_code == 200, res.body + tools = res.body + # Should have builtin tools but no MCP tools from the bad server + mcp_tools = [t for t in tools if t.get("name", "").startswith("nonexistent_")] + assert len(mcp_tools) == 0, f"Expected no nonexistent_ tools, got {mcp_tools}" + finally: + server.stop() + + +def test_mcp_multiple_servers(): + """Multiple MCP servers can be configured simultaneously.""" + global server + mcp_json = _mcp_config_json({ + "echo": { + "command": sys.executable, + "args": [MCP_ECHO_SERVER], + }, + "echo2": { + "command": sys.executable, + "args": [MCP_ECHO_SERVER], + } + }) + server = _start_server_with_mcp(mcp_json) + + try: + res = server.make_request("GET", "/tools") + assert res.status_code == 200, res.body + + tools = res.body + + def get_tool_name(t): + return t.get("tool", "") or t.get("definition", {}).get("function", {}).get("name", "") + + echo_tools = [t for t in tools if get_tool_name(t).startswith("echo_")] + echo2_tools = [t for t in tools if get_tool_name(t).startswith("echo2_")] + + assert len(echo_tools) >= 2, f"Expected echo_ tools, got {echo_tools}" + assert len(echo2_tools) >= 2, f"Expected echo2_ tools, got {echo2_tools}" + finally: + server.stop() + + +def test_mcp_tools_not_listed_when_not_configured(): + """Without MCP config, no MCP tools should appear.""" + global server + server = ServerPreset.router() + server.server_tools = "all" + server.no_ui = True + server.server_port = 8085 + server.start() + + try: + res = server.make_request("GET", "/tools") + assert res.status_code == 200, res.body + + tools = res.body + + def get_tool_name(t): + return t.get("tool", "") or t.get("definition", {}).get("function", {}).get("name", "") + + # Should only have builtin tools, no server: prefixed tools + mcp_tools = [t for t in tools if ":" in get_tool_name(t)] + assert len(mcp_tools) == 0, f"Expected no MCP tools, got {mcp_tools}" + finally: + server.stop() + + +def test_mcp_fail_once_tool_eventual_success(): + """Test that a tool that fails once eventually succeeds (tests instance respawn).""" + global server + mcp_json = _mcp_config_json({ + "echo": { + "command": sys.executable, + "args": [MCP_ECHO_SERVER], + } + }) + server = _start_server_with_mcp(mcp_json) + + try: + # First call should succeed (warmup already spawned and shut down the instance, + # but the first actual tool call will spawn a fresh instance) + res = server.make_request("POST", "/tools", data={ + "tool": "echo_fail_once", + "params": {} + }) + # It might fail on first call if the warmup instance was shut down + # and a new instance is spawned. The fail_once state is per-process, + # so a fresh process will fail once then succeed. + # Actually, warmup spawns, lists, then shuts down. So the first tool call + # spawns a new process which will fail once. + assert res.status_code in (200, 500), res.body + finally: + server.stop() + + +def test_mcp_tools_via_json_config_file(): + """Test that --mcp-servers-config (file) works as well as --mcp-servers-json.""" + global server + config = { + "mcpServers": { + "echo": { + "command": sys.executable, + "args": [MCP_ECHO_SERVER], + } + } + } + + with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: + json.dump(config, f) + config_path = f.name + + try: + server = ServerPreset.router() + server.server_tools = "all" + server.no_ui = True + server.server_port = 8085 + server.mcp_servers_config = config_path + server.start() + + res = server.make_request("GET", "/tools") + assert res.status_code == 200, res.body + + tools = res.body + + def get_tool_name(t): + return t.get("tool", "") or t.get("definition", {}).get("function", {}).get("name", "") + + mcp_tools = [t for t in tools if get_tool_name(t).startswith("echo_")] + assert len(mcp_tools) >= 2, f"Expected echo_ tools, got {mcp_tools}" + finally: + os.unlink(config_path) + server.stop() + + +def test_mcp_tools_slot_independent(): + """MCP tools should work without any slot concept; /tools is slot-independent.""" + global server + mcp_json = _mcp_config_json({ + "echo": { + "command": sys.executable, + "args": [MCP_ECHO_SERVER], + } + }) + server = _start_server_with_mcp(mcp_json) + + try: + # Call /tools without any slot binding - should succeed + res = server.make_request("POST", "/tools", data={ + "tool": "echo_echo", + "params": {"message": "hello"} + }) + assert res.status_code == 200, res.body + body = res.body + assert "error" not in body, body + finally: + server.stop() + + +def test_mcp_concurrent_tool_calls(): + """Concurrent POST /tools to same MCP server should all succeed.""" + global server + mcp_json = _mcp_config_json({ + "echo": { + "command": sys.executable, + "args": [MCP_ECHO_SERVER], + } + }) + server = _start_server_with_mcp(mcp_json) + + try: + def call_tool(): + return server.make_request("POST", "/tools", data={ + "tool": "echo_echo", + "params": {"message": "hi"} + }) + + with ThreadPoolExecutor(max_workers=10) as executor: + futures = [executor.submit(call_tool) for _ in range(10)] + results = [f.result() for f in futures] + + for res in results: + assert res.status_code == 200, res.body + assert "error" not in res.body, res.body + finally: + server.stop() + + +def test_mcp_tool_timeout(): + """Tool call should timeout if MCP server is too slow.""" + global server + MCP_SLOW_SERVER = os.path.join(FIXTURES_DIR, "mcp_slow_server.py") + mcp_json = _mcp_config_json({ + "slow": { + "command": sys.executable, + "args": [MCP_SLOW_SERVER, "--delay", "5"], + "timeout_ms": 500 + } + }) + server = _start_server_with_mcp(mcp_json) + + try: + res = server.make_request("POST", "/tools", data={ + "tool": "slow_sleep", + "params": {"seconds": 5} + }) + assert res.status_code == 200, res.body + body = res.body + assert "error" in body, body + finally: + server.stop() + + +def test_mcp_warmup_partial_failure(): + """Good server's tools should appear even if bad server fails warmup.""" + global server + mcp_json = _mcp_config_json({ + "good": { + "command": sys.executable, + "args": [MCP_ECHO_SERVER], + }, + "bad": { + "command": "nonexistent", + "args": [] + } + }) + server = _start_server_with_mcp(mcp_json) + + try: + res = server.make_request("GET", "/tools") + assert res.status_code == 200, res.body + tools = res.body + + def get_tool_name(t): + return t.get("tool", "") or t.get("definition", {}).get("function", {}).get("name", "") + + # good server tools should be present + assert any("good_" in get_tool_name(t) for t in tools), f"Expected good: tools in {tools}" + finally: + server.stop() + + +def test_mcp_notification_during_request(): + """Notification during request should not be returned as response.""" + global server + MCP_MALFORMED_SERVER = os.path.join(FIXTURES_DIR, "mcp_malformed_server.py") + mcp_json = _mcp_config_json({ + "notifying": { + "command": sys.executable, + "args": [MCP_MALFORMED_SERVER], + } + }) + server = _start_server_with_mcp(mcp_json) + + try: + res = server.make_request("POST", "/tools", data={ + "tool": "notifying_echo", + "params": {"message": "hi"} + }) + assert res.status_code == 200, res.body + body = res.body + assert "error" not in body, body + finally: + server.stop() + + +def test_mcp_instance_respawn_after_crash(): + """Tool call after process crash should respawn and succeed.""" + global server + MCP_CRASH_SERVER = os.path.join(FIXTURES_DIR, "mcp_crash_server.py") + mcp_json = _mcp_config_json({ + "crash": { + "command": sys.executable, + "args": [MCP_CRASH_SERVER], + } + }) + server = _start_server_with_mcp(mcp_json) + + try: + # First call succeeds + res1 = server.make_request("POST", "/tools", data={ + "tool": "crash_echo", + "params": {"message": "hi"} + }) + assert res1.status_code == 200, res1.body + assert "error" not in res1.body, res1.body + + # Second call should also succeed (respawned instance) + res2 = server.make_request("POST", "/tools", data={ + "tool": "crash_echo", + "params": {"message": "hi2"} + }) + assert res2.status_code == 200, res2.body + assert "error" not in res2.body, res2.body + finally: + server.stop() + + + + +def test_mcp_fail_once_eventual_success_verified(): + """Verify that fail_once tool eventually succeeds after respawn.""" + global server + mcp_json = _mcp_config_json({ + "echo": { + "command": sys.executable, + "args": [MCP_ECHO_SERVER], + } + }) + server = _start_server_with_mcp(mcp_json) + + try: + # First call may fail (fresh process) + res1 = server.make_request("POST", "/tools", data={ + "tool": "echo_fail_once", + "params": {} + }) + # Second call should succeed + res2 = server.make_request("POST", "/tools", data={ + "tool": "echo_fail_once", + "params": {} + }) + assert res2.status_code == 200, res2.body + assert "error" not in res2.body, res2.body + finally: + server.stop() + + +def test_mcp_config_file_errors(): + """Invalid JSON config and missing file should cause server to fail to start.""" + # Invalid JSON - server should fail to start + server = ServerPreset.router() + server.server_tools = "all" + server.no_ui = True + server.server_port = 8085 + server.mcp_servers_json = "not valid json" + try: + server.start() + assert False, "Server should not have started with invalid MCP JSON config" + except RuntimeError: + pass # Expected: server process dies due to bad config + + # Missing file - server should fail to start + server = ServerPreset.router() + server.server_tools = "all" + server.no_ui = True + server.server_port = 8085 + server.mcp_servers_config = "/nonexistent/path.json" + try: + server.start() + assert False, "Server should not have started with missing config file" + except RuntimeError: + pass # Expected: server process dies due to missing config + + +def test_mcp_empty_tool_list(): + """MCP server reporting zero tools should result in empty tool list.""" + global server + # Create a minimal server that returns empty tools list + empty_server = os.path.join(FIXTURES_DIR, "_empty_mcp_server.py") + with open(empty_server, "w") as f: + f.write('''#!/usr/bin/env python3 +import json, sys, os +def main(): + sys.stdout = os.fdopen(sys.stdout.fileno(), "w", buffering=1) + for line in sys.stdin: + line = line.strip() + if not line: continue + try: request = json.loads(line) + except: continue + method = request.get("method") + req_id = request.get("id") + if method == "initialize": + resp = {"jsonrpc": "2.0", "id": req_id, "result": {"protocolVersion": "2024-11-05", "capabilities": {"tools": {}}, "serverInfo": {"name": "empty", "version": "1.0"}}} + elif method == "tools/list": + resp = {"jsonrpc": "2.0", "id": req_id, "result": {"tools": []}} + else: + resp = {"jsonrpc": "2.0", "id": req_id, "error": {"code": -32601, "message": "Method not found"}} + sys.stdout.write(json.dumps(resp) + "\\n") + sys.stdout.flush() +if __name__ == "__main__": + main() +''') + try: + mcp_json = _mcp_config_json({ + "empty": { + "command": sys.executable, + "args": [empty_server], + } + }) + server = _start_server_with_mcp(mcp_json) + res = server.make_request("GET", "/tools") + assert res.status_code == 200, res.body + tools = res.body + def get_tool_name(t): + return t.get("tool", "") or t.get("definition", {}).get("function", {}).get("name", "") + mcp_tools = [t for t in tools if get_tool_name(t).startswith("empty:")] + assert len(mcp_tools) == 0, f"Expected no empty: tools, got {mcp_tools}" + finally: + os.unlink(empty_server) + server.stop() + + +def test_mcp_rapid_succession_calls(): + """Many rapid calls should increment next_id correctly and correlate responses.""" + global server + mcp_json = _mcp_config_json({ + "echo": { + "command": sys.executable, + "args": [MCP_ECHO_SERVER], + } + }) + server = _start_server_with_mcp(mcp_json) + + try: + for i in range(20): + res = server.make_request("POST", "/tools", data={ + "tool": "echo_echo", + "params": {"message": f"msg{i}"} + }) + assert res.status_code == 200, res.body + assert "error" not in res.body, res.body + finally: + server.stop() + + +def test_mcp_notification_burst(): + """Notification + response in a single write() with no flush should not strand the response.""" + global server + MCP_BURST_SERVER = os.path.join(FIXTURES_DIR, "mcp_burst_server.py") + mcp_json = _mcp_config_json({ + "burst": { + "command": sys.executable, + "args": [MCP_BURST_SERVER], + } + }) + server = _start_server_with_mcp(mcp_json) + + try: + res = server.make_request("POST", "/tools", data={ + "tool": "burst_echo", + "params": {"message": "burst test"} + }) + assert res.status_code == 200, res.body + body = res.body + assert "error" not in body, body + finally: + server.stop() + + +def test_mcp_tool_definition_shape_via_chat_completions(): + """MCP tool definitions returned by GET /tools should have the correct shape for chat/completions.""" + global server + mcp_json = _mcp_config_json({ + "echo": { + "command": sys.executable, + "args": [MCP_ECHO_SERVER], + } + }) + server = _start_server_with_mcp(mcp_json) + + try: + # Get MCP tool definitions + res = server.make_request("GET", "/tools") + assert res.status_code == 200, res.body + tools = res.body + + def get_tool_name(t): + return t.get("tool", "") or t.get("definition", {}).get("function", {}).get("name", "") + + echo_tools = [t for t in tools if get_tool_name(t).startswith("echo_")] + assert len(echo_tools) >= 2, f"Expected echo_ tools, got {echo_tools}" + + echo_tool = next(t for t in echo_tools if get_tool_name(t) == "echo_echo") + definition = echo_tool.get("definition", echo_tool) + + # Verify the definition has the standard function-calling shape + assert definition.get("type") == "function", f"Expected type=function, got {definition.get('type')}" + func = definition.get("function", {}) + assert "name" in func, "Missing function.name" + assert "description" in func, "Missing function.description" + assert "parameters" in func, f"Missing function.parameters, got keys: {list(func.keys())}" + params = func["parameters"] + assert params.get("type") == "object", f"Expected parameters.type=object, got {params.get('type')}" + assert "properties" in params, "Missing parameters.properties" + finally: + server.stop() + + +def test_mcp_slow_tool_call_slot_release(): + """A slow tool call should not stall server shutdown for the full I/O timeout.""" + global server + MCP_SLOW_SERVER = os.path.join(FIXTURES_DIR, "mcp_slow_server.py") + mcp_json = _mcp_config_json({ + "slow": { + "command": sys.executable, + "args": [MCP_SLOW_SERVER, "--delay", "10"], + "timeout_ms": 30000 + } + }) + server = _start_server_with_mcp(mcp_json) + + try: + # Start a slow tool call in a background thread + def slow_call(): + return server.make_request("POST", "/tools", data={ + "tool": "slow_sleep", + "params": {"seconds": 10} + }) + + with ThreadPoolExecutor(max_workers=1) as executor: + future = executor.submit(slow_call) + + # Wait a moment for the call to start + time.sleep(2) + + # Stop the server while the tool call is in progress. + # With global MCP instances, close_all() is called explicitly at shutdown + # (not from slot release), so shutdown should complete promptly. + start_time = time.time() + server.stop() + elapsed = time.time() - start_time + + # The server should stop quickly, not wait for the full 30s I/O timeout. + # With the terminating flag, send_rpc() bails out within one select() + # slice (~50ms). This threshold MUST stay below the 5s force-kill + # fallback in ServerProcess.stop(): without the flag, shutdown stalls + # on the instance mutex and only completes when stop() sends SIGKILL + # at ~5s -- which any threshold above 5 would still accept. + assert elapsed < 3, f"Server stop took {elapsed:.1f}s, expected < 3s" + + # Wait for the future to complete (it will get an error response or timeout) + try: + res = future.result(timeout=5) + # If we got a response, it should be an error since the server stopped + if hasattr(res, 'status_code'): + assert res.status_code in (200, 500, 502, 503, 504), f"Unexpected status: {res.status_code}" + except Exception: + # Thread may have raised due to connection error - that's acceptable + pass + finally: + server.stop() + + +def test_mcp_grandchild_holding_pipes_does_not_deadlock(): + """An MCP server that leaves a grandchild inheriting its stdout/stderr must not deadlock + teardown. + + subprocess_terminate() only SIGKILLs the direct MCP child, so the inherited pipe write ends + stay open and a blocking read on them would never see EOF. That hung both warmup shutdown + (the server would never reach "ready") and process shutdown. The polled, running-aware reader + must exit regardless, so the server both starts and stops promptly here. + """ + global server + MCP_GRANDCHILD_SERVER = os.path.join(FIXTURES_DIR, "mcp_grandchild_server.py") + mcp_json = _mcp_config_json({ + "gc": { + "command": sys.executable, + "args": [MCP_GRANDCHILD_SERVER], + } + }) + + # If warmup teardown deadlocked, the server would never become ready and start() would time out. + server = _start_server_with_mcp(mcp_json) + + try: + # invoking the tool spawns a live transport whose reader thread holds the inherited pipe + res = server.make_request("POST", "/tools", data={ + "tool": "gc_echo", + "params": {"message": "hello"} + }) + assert res.status_code == 200, res.body + assert "error" not in res.body, res.body + + # shutdown must be prompt: a deadlocked reader-join would stall until the 5s SIGKILL + # fallback in ServerProcess.stop(), so the threshold has to stay below that + start = time.time() + server.stop() + elapsed = time.time() - start + assert elapsed < 3, f"server shutdown took {elapsed:.1f}s (expected < 3s) — teardown likely deadlocked" + finally: + server.stop() diff --git a/tools/server/tests/utils.py b/tools/server/tests/utils.py index 5d5c873ac..ae56bc70a 100644 --- a/tools/server/tests/utils.py +++ b/tools/server/tests/utils.py @@ -115,6 +115,8 @@ class ServerProcess: backend_sampling: bool = False gcp_compat: bool = False server_tools: str | None = None + mcp_servers_config: str | None = None + mcp_servers_json: str | None = None cors_origins: str | None = None # session variables @@ -265,6 +267,10 @@ class ServerProcess: server_args.append("--ui-mcp-proxy") if self.server_tools: server_args.extend(["--tools", self.server_tools]) + if self.mcp_servers_config: + server_args.extend(["--mcp-servers-config", self.mcp_servers_config]) + if self.mcp_servers_json: + server_args.extend(["--mcp-servers-json", self.mcp_servers_json]) if self.backend_sampling: server_args.append("--backend_sampling") if self.gcp_compat: