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";
}
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"},
// 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
server-stream.h
server-tools.cpp
server-tools.h
+ server-mcp.cpp
+ server-mcp.h
server-schema.cpp
server-schema.h
)
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
- `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:
#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;
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;
+ }
+};
--- /dev/null
+#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;
+}
+
--- /dev/null
+#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);
+};
// 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); });
) {
// 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
#include <ctime>
#include <atomic>
#include <cstring>
-#include <climits>
#include <algorithm>
#include <unordered_set>
#include <functional>
+#include <memory>
namespace fs = std::filesystem;
return {
{"display_name", display_name},
{"tool", name},
- {"type", "builtin"},
+ {"type", type()},
{"permissions", json{
{"write", permission_write}
}},
}
};
+//
+// 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) {
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();
}
}
+ // 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 {
#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;
virtual ~server_tool() = default;
virtual json get_definition() const = 0;
+ virtual std::string type() const { return "builtin"; }
struct stream {
server_response & qr;
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;
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;
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;
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));
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();
models_routes->stopping.store(true); // maybe redundant, but just to be safe
models_routes->models.unload_all();
}
+ mcp_mgr.shutdown();
llama_backend_free();
};
// 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();
};
SRV_INF("%s", "model loaded\n");
shutdown_handler = [&](int) {
+ mcp_mgr.shutdown();
// this will unblock start_loop()
ctx_server.terminate();
};
--- /dev/null
+#!/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()
--- /dev/null
+#!/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()
--- /dev/null
+#!/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()
--- /dev/null
+#!/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()
--- /dev/null
+#!/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()
--- /dev/null
+#!/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()
--- /dev/null
+#!/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()
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
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:
overview / pkg / ggml / sources / llama.cpp / commitdiff