server: support MCP stdio (#26062)
* move server_pipe to common * init impl * vendor: update subprocess.h * add server_mcp_stdio * stderr drain * server_mcp_transport * server_mcp_stdio is now framing-only, no json * internal/mcp-stdio: integration + tests + fixes (#26075) * server-mcp: harden transport and wire up the tool integration Builds on the transport/manager architecture (server_mcp_transport + server_pipe) with the hardening and integration the draft did not yet have. Hardening: * Reader and stderr pumps are polled (running-aware) instead of blocking on a read that only ends at EOF. subprocess_terminate() SIGKILLs only the direct child, so a grandchild the MCP server spawned that inherited the pipe would otherwise keep the write end open and hang teardown (both warmup shutdown at startup and process shutdown). The writer is likewise non-blocking + polled. * Windows: resolve the command through PATHEXT so "npx" (npm ships npx.cmd, never npx.exe) spawns, matching POSIX's PATH search; and enumerate the parent environment as UTF-8 (GetEnvironmentStringsW) instead of the active code page. * server_pipe gains an opt-in max_size (default unbounded, so the router's streaming use is unchanged); the MCP reply queue uses it so a server that streams unsolicited notifications between requests cannot grow it without bound. Integration: * --mcp-servers-config / --mcp-servers-json flags; enabling MCP restricts default CORS to localhost, same as --tools. * MCP tools are exposed through /tools (and chat-completions) as <server>_<tool>, skipping names that collide with a built-in or another MCP tool. * Manager lifecycle wired into llama_server(): warmup at start, shutdown() from the signal handler before the HTTP server drains, blocking teardown in clean_up(). * SIGPIPE ignored so a child dying mid-write yields EPIPE rather than killing us. Assisted-By: Claude Opus 4.8 <noreply@anthropic.com> * server-mcp: add MCP test suite with grandchild deadlock regression test 21 tests over the /tools endpoint: tool discovery/invocation, timeouts, crash recovery and respawn cooldown, warmup partial failure, malformed and batched notification+response output, tool-definition shape, and prompt shutdown during a slow call. The last test spawns an MCP server that leaves a grandchild inheriting its stdout/stderr and asserts the server both starts and stops promptly. Verified it fails (5s SIGKILL fallback on a deadlocked reader-join) when the pump is made to ignore the running flag, and passes with the polled reader. Assisted-By: Claude Opus 4.8 <noreply@anthropic.com> * clean up * clean up 2 * even stricter life cycle * nits * nits 2 --------- Co-authored-by: Xuan Son Nguyen <son@huggingface.co> * fix some edge cases * fix last_error data race * fix response schema + docs * server: fix MCP zombie leak and timeout-induced transport teardown join_pumps() never reaped the child, leaking one zombie per spawn: call subprocess_join() before subprocess_destroy(). A per-call timeout permanently closed from_server and got a healthy transport evicted: add close_on_stop to server_pipe::read() and pass false from send_rpc(), where should_stop is a per-request deadline and a late reply is already skipped on id mismatch. Also drop the unreachable disconnect cancellation in server_mcp_tool::invoke(): support_stream is false, st is always null. (cherry picked from commit e6de1ec043174fd0570b1e60d47f06c7c19d620d) Assisted-by: Claude Opus 4.8 * server: make MCP test fixtures JSON-RPC 2.0 compliant Add the missing notification guard to mcp_malformed_server.py and mcp_burst_server.py (the latter treated id 0 as a notification and replied to unknown ones; its notification table is now unused). Return -32602 instead of -32601 for unknown tools: tools/call is a valid method, the tool name is the invalid parameter. Also fix the test module docstring: tools are named <server>_<tool>. (cherry picked from commit 74a08e8c311dabf3b49d06cc6d754b0097ae7a38) Assisted-by: Claude Opus 4.8 --------- Co-authored-by: Piotr Wilkin (ilintar) <piotr.wilkin@syndatis.com> Co-authored-by: Pascal <admin@serveurperso.com>
This commit is contained in:
Xuan-Son Nguyen
co-authored by
Piotr Wilkin
Pascal
Piotr Wilkin
Pascal
parent
355303edab
commit
20455a4ad3
+19
-2
@@ -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"},
|
||||
|
||||
@@ -668,6 +668,10 @@ struct common_params {
|
||||
// enable built-in tools
|
||||
std::vector<std::string> 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
|
||||
|
||||
@@ -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
|
||||
)
|
||||
|
||||
@@ -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:
|
||||
|
||||
|
||||
@@ -9,9 +9,15 @@
|
||||
#define JSON_ASSERT GGML_ASSERT
|
||||
#include <nlohmann/json.hpp>
|
||||
|
||||
#include <atomic>
|
||||
#include <chrono>
|
||||
#include <condition_variable>
|
||||
#include <cinttypes>
|
||||
#include <functional>
|
||||
#include <mutex>
|
||||
#include <queue>
|
||||
#include <string>
|
||||
#include <vector>
|
||||
#include <cinttypes>
|
||||
|
||||
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<typename T>
|
||||
struct server_pipe {
|
||||
std::mutex mutex;
|
||||
std::condition_variable cv;
|
||||
std::queue<T> queue;
|
||||
std::atomic<bool> writer_closed{false};
|
||||
std::atomic<bool> 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<bool()> & should_stop, bool close_on_stop = true) {
|
||||
std::unique_lock<std::mutex> 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<std::mutex> 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;
|
||||
}
|
||||
};
|
||||
|
||||
@@ -0,0 +1,836 @@
|
||||
#include "server-mcp.h"
|
||||
|
||||
#include <sheredom/subprocess.h>
|
||||
|
||||
#include <atomic>
|
||||
#include <chrono>
|
||||
#include <cstdio>
|
||||
#include <fstream>
|
||||
#include <functional>
|
||||
#include <sstream>
|
||||
#include <thread>
|
||||
|
||||
#if defined(_WIN32)
|
||||
# include <io.h>
|
||||
# include <windows.h>
|
||||
#else
|
||||
# include <errno.h>
|
||||
# include <fcntl.h>
|
||||
# include <poll.h>
|
||||
# include <unistd.h>
|
||||
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<bool> & running,
|
||||
const std::function<bool(std::string &&)> & 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> 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> server_mcp_server_config::parse_cursor_format(const json & j) {
|
||||
std::vector<server_mcp_server_config> 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<std::string>());
|
||||
}
|
||||
}
|
||||
if (cfg.contains("env") && cfg.at("env").is_object()) {
|
||||
for (const auto & [k, v] : cfg.at("env").items()) {
|
||||
sc.env[k] = v.get<std::string>();
|
||||
}
|
||||
}
|
||||
|
||||
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<std::string>();
|
||||
}
|
||||
}
|
||||
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<bool()> & 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<bool()> & 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_tool_def> server_mcp_transport::list_tools(const std::function<bool()> & should_stop) {
|
||||
std::lock_guard<std::mutex> 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<bool()> & should_stop) {
|
||||
std::lock_guard<std::mutex> 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<std::string> mcp_parent_env() {
|
||||
std::vector<std::string> 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<std::string> mcp_build_env(const std::map<std::string, std::string> & overrides) {
|
||||
std::vector<std::string> 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<std::string> 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<std::string> envp_s;
|
||||
if (config.env.empty()) {
|
||||
options |= subprocess_option_inherit_environment;
|
||||
} else {
|
||||
envp_s = mcp_build_env(config.env);
|
||||
}
|
||||
|
||||
auto to_ptrs = [](std::vector<std::string> & v) {
|
||||
std::vector<const char *> 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<process_handle>();
|
||||
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<std::mutex> lock(rpc_mutex); // last_error is written by send_rpc's callers
|
||||
out = last_error;
|
||||
}
|
||||
std::lock_guard<std::mutex> 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<bool> & 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<std::mutex> 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<std::shared_ptr<server_mcp_transport>> to_close;
|
||||
{
|
||||
std::lock_guard<std::mutex> 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_transport> server_mcp::create_transport(const server_mcp_server_config & cfg) {
|
||||
return std::make_shared<server_mcp_stdio>(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<server_mcp_tool_def> 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<std::mutex> lock(mutex);
|
||||
registry.swap(discovered);
|
||||
}
|
||||
|
||||
std::vector<server_mcp_tool_def> server_mcp::list_tools() const {
|
||||
std::lock_guard<std::mutex> 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<bool()> & 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_transport> server_mcp::get_or_create(const std::string & name) {
|
||||
std::vector<std::shared_ptr<server_mcp_transport>> to_close; // closed after unlock
|
||||
std::shared_ptr<server_mcp_transport> result;
|
||||
|
||||
{
|
||||
std::lock_guard<std::mutex> 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;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,176 @@
|
||||
#pragma once
|
||||
|
||||
#include "server-common.h"
|
||||
|
||||
#include <atomic>
|
||||
#include <chrono>
|
||||
#include <functional>
|
||||
#include <map>
|
||||
#include <memory>
|
||||
#include <mutex>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
#include <vector>
|
||||
|
||||
//
|
||||
// Configuration (Cursor-compatible "mcpServers" JSON)
|
||||
//
|
||||
|
||||
struct server_mcp_server_config {
|
||||
std::string name; // config key, e.g. "filesystem"
|
||||
std::string command;
|
||||
std::vector<std::string> args;
|
||||
std::map<std::string, std::string> 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<server_mcp_server_config> parse_from_json(const std::string & json_str);
|
||||
static std::vector<server_mcp_server_config> 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 "<server>_" 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<std::string> to_server; // serialized messages we send to the server
|
||||
server_pipe<std::string> 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<std::mutex> lock(rpc_mutex);
|
||||
return last_error;
|
||||
}
|
||||
|
||||
std::vector<server_mcp_tool_def> list_tools(const std::function<bool()> & should_stop);
|
||||
|
||||
json call_tool(const std::string & tool_name,
|
||||
const json & arguments,
|
||||
const std::function<bool()> & 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<server_mcp_tool_def> tools;
|
||||
|
||||
// both assume rpc_mutex is already held by the public caller
|
||||
bool ensure_init(const std::function<bool()> & should_stop); // initialize handshake, once
|
||||
json send_rpc(const json & request, const std::function<bool()> & 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 <windows.h> stays out of this header
|
||||
struct process_handle;
|
||||
std::unique_ptr<process_handle> 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<bool> 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<server_mcp_tool_def> 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<bool()> & 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<server_mcp_server_config> 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<std::string, std::shared_ptr<server_mcp_transport>> transports;
|
||||
std::map<std::string, std::chrono::steady_clock::time_point> dead_servers; // spawn-failure cooldown
|
||||
std::vector<server_mcp_tool_def> registry;
|
||||
|
||||
std::atomic<bool> 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<server_mcp_transport> create_transport(const server_mcp_server_config & cfg);
|
||||
|
||||
// nullptr during cooldown or shutdown
|
||||
std::shared_ptr<server_mcp_transport> get_or_create(const std::string & name);
|
||||
};
|
||||
@@ -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<typename T>
|
||||
struct pipe_t {
|
||||
std::mutex mutex;
|
||||
std::condition_variable cv;
|
||||
std::queue<T> queue;
|
||||
std::atomic<bool> writer_closed{false};
|
||||
std::atomic<bool> 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<bool()> & should_stop) {
|
||||
std::unique_lock<std::mutex> 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<std::mutex> 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<httplib::ClientImpl>(host, port);
|
||||
auto pipe = std::make_shared<pipe_t<msg_t>>();
|
||||
auto pipe = std::make_shared<server_pipe<msg_t>>();
|
||||
|
||||
if (scheme == "https") {
|
||||
#ifdef CPPHTTPLIB_OPENSSL_SUPPORT
|
||||
|
||||
@@ -10,10 +10,10 @@
|
||||
#include <ctime>
|
||||
#include <atomic>
|
||||
#include <cstring>
|
||||
#include <climits>
|
||||
#include <algorithm>
|
||||
#include <unordered_set>
|
||||
#include <functional>
|
||||
#include <memory>
|
||||
|
||||
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<std::unique_ptr<server_tool>> & tools, const std::string & name, bool require_stream) {
|
||||
for (auto & t : tools) {
|
||||
if (t->name == name) {
|
||||
@@ -1157,7 +1200,8 @@ static std::vector<std::unique_ptr<server_tool>> build_tools() {
|
||||
return tools;
|
||||
}
|
||||
|
||||
void server_tools::setup(const std::vector<std::string> & enabled_tools) {
|
||||
void server_tools::setup(const std::vector<std::string> & enabled_tools,
|
||||
server_mcp & mcp_mgr) {
|
||||
if (!enabled_tools.empty()) {
|
||||
std::unordered_set<std::string> enabled_set(enabled_tools.begin(), enabled_tools.end());
|
||||
auto all_tools = build_tools();
|
||||
@@ -1188,6 +1232,29 @@ void server_tools::setup(const std::vector<std::string> & enabled_tools) {
|
||||
}
|
||||
}
|
||||
|
||||
// append MCP tools, skipping any that collide with a built-in or another MCP tool of the same "<server>_<tool>" name
|
||||
if (!mcp_mgr.empty()) {
|
||||
std::unordered_set<std::string> 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<server_mcp_tool>(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<server_http_res>();
|
||||
try {
|
||||
|
||||
@@ -3,9 +3,11 @@
|
||||
#include "server-common.h"
|
||||
#include "server-http.h"
|
||||
#include "server-queue.h"
|
||||
#include "server-mcp.h"
|
||||
|
||||
#include <atomic>
|
||||
#include <functional>
|
||||
#include <memory>
|
||||
|
||||
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<int> res_id{0};
|
||||
|
||||
void setup(const std::vector<std::string> & enabled_tools);
|
||||
void setup(const std::vector<std::string> & enabled_tools,
|
||||
server_mcp & mcp_mgr);
|
||||
|
||||
server_http_context::handler_t handle_get;
|
||||
server_http_context::handler_t handle_post;
|
||||
|
||||
+29
-6
@@ -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();
|
||||
};
|
||||
|
||||
+118
@@ -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()
|
||||
+114
@@ -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()
|
||||
+164
@@ -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()
|
||||
@@ -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()
|
||||
+113
@@ -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()
|
||||
+132
@@ -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()
|
||||
@@ -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 <server>_<tool> 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()
|
||||
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user