Felipe97/llama-cpp-compiled
01.1k
1#include "server-mcp.h"2 3#include "subproc.h"4 5#include <atomic>6#include <chrono>7#include <cstdio>8#include <fstream>9#include <functional>10#include <sstream>11#include <thread>12 13#if defined(_WIN32)14# include <io.h>15# include <windows.h>16#else17# include <errno.h>18# include <fcntl.h>19# include <poll.h>20# include <unistd.h>21extern char ** environ;22#endif23 24// read NDJSON lines from a child pipe, calling on_line per line until `running` clears, EOF/error, or on_line returns false.25// 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.26static void mcp_pump_ndjson(FILE * f, std::atomic<bool> & running,27 const std::function<bool(std::string &&)> & on_line) {28 if (!f) {29 return;30 }31 const int poll_ms = 50;32 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 bound33#if defined(_WIN32)34 HANDLE h = (HANDLE) _get_osfhandle(_fileno(f));35#else36 int fd = fileno(f);37 int fl = fcntl(fd, F_GETFL, 0);38 if (fl >= 0) {39 fcntl(fd, F_SETFL, fl | O_NONBLOCK);40 }41#endif42 std::string buf;43 bool skipping = false; // discarding an over-long line until its terminating newline44 char chunk[4096];45 while (running.load()) {46 size_t n = 0;47#if defined(_WIN32)48 DWORD avail = 0;49 if (!PeekNamedPipe(h, NULL, 0, NULL, &avail, NULL)) {50 break; // pipe broken / child gone51 }52 if (avail == 0) {53 std::this_thread::sleep_for(std::chrono::milliseconds(poll_ms));54 continue;55 }56 DWORD to_read = avail < (DWORD) sizeof(chunk) ? avail : (DWORD) sizeof(chunk);57 DWORD got = 0;58 if (!ReadFile(h, chunk, to_read, &got, NULL) || got == 0) {59 break;60 }61 n = (size_t) got;62#else63 struct pollfd pfd;64 pfd.fd = fd;65 pfd.events = POLLIN;66 pfd.revents = 0;67 int pr = poll(&pfd, 1, poll_ms);68 if (pr < 0) {69 if (errno == EINTR) {70 continue;71 }72 break;73 }74 if (pr == 0) {75 continue; // timeout -> re-check running76 }77 if (pfd.revents & (POLLERR | POLLNVAL)) {78 break;79 }80 ssize_t r = read(fd, chunk, sizeof(chunk));81 if (r < 0) {82 if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) {83 continue;84 }85 break;86 }87 if (r == 0) {88 break; // EOF: child (and any pipe writers) closed the stream89 }90 n = (size_t) r;91#endif92 buf.append(chunk, n);93 94 // resync after an over-long, unterminated line: discard bytes until the next newline95 if (skipping) {96 size_t nl = buf.find('\n');97 if (nl == std::string::npos) {98 if (buf.size() > max_line) {99 buf.clear(); // stay bounded while waiting for a terminator100 }101 continue;102 }103 buf.erase(0, nl + 1);104 skipping = false;105 }106 107 size_t pos;108 while ((pos = buf.find('\n')) != std::string::npos) {109 std::string line = buf.substr(0, pos);110 buf.erase(0, pos + 1);111 if (!line.empty() && line.back() == '\r') {112 line.pop_back();113 }114 if (line.empty()) {115 continue;116 }117 if (!on_line(std::move(line))) {118 return;119 }120 }121 122 // a partial line already larger than the cap and still no newline: drop it to avoid unbounded growth123 if (buf.size() > max_line) {124 SRV_WRN("MCP: dropping oversized line (> %zu bytes) from child pipe\n", max_line);125 buf.clear();126 skipping = true;127 }128 }129}130 131//132// server_mcp_server_config133//134 135std::vector<server_mcp_server_config> server_mcp_server_config::parse_from_json(const std::string & json_str) {136 return parse_cursor_format(json::parse(json_str));137}138 139std::vector<server_mcp_server_config> server_mcp_server_config::parse_cursor_format(const json & j) {140 std::vector<server_mcp_server_config> result;141 142 if (!j.contains("mcpServers") || !j.at("mcpServers").is_object()) {143 return result;144 }145 146 for (const auto & [name, cfg] : j.at("mcpServers").items()) {147 server_mcp_server_config sc;148 sc.name = name;149 sc.command = cfg.value("command", std::string());150 sc.cwd = cfg.value("cwd", std::string());151 sc.timeout_ms = cfg.value("timeout_ms", sc.timeout_ms);152 153 if (cfg.contains("args") && cfg.at("args").is_array()) {154 for (const auto & a : cfg.at("args")) {155 sc.args.push_back(a.get<std::string>());156 }157 }158 if (cfg.contains("env") && cfg.at("env").is_object()) {159 for (const auto & [k, v] : cfg.at("env").items()) {160 sc.env[k] = v.get<std::string>();161 }162 }163 164 if (sc.command.empty()) {165 SRV_WRN("MCP server '%s' has no command, skipping\n", name.c_str());166 continue;167 }168 result.push_back(std::move(sc));169 }170 171 return result;172}173 174 175//176// server_mcp_transport177//178 179static constexpr const char * MCP_PROTOCOL_VERSION = "2024-11-05";180 181static std::string rpc_error_message(const json & resp) {182 if (resp.contains("error")) {183 const json & e = resp.at("error");184 if (e.is_object()) {185 return e.value("message", "unknown error");186 }187 if (e.is_string()) {188 return e.get<std::string>();189 }190 }191 return "unknown error";192}193 194// normalize an MCP tools/call result to the /tools contract (see README-dev.md):195// concat text parts of result.content[], and surface an isError result196static json mcp_result_to_response(const json & result) {197 std::string text;198 if (result.contains("content") && result.at("content").is_array()) {199 for (const auto & part : result.at("content")) {200 if (part.is_object() && part.value("type", "") == "text") {201 if (!text.empty()) {202 text += "\n";203 }204 text += part.value("text", "");205 }206 }207 }208 if (result.is_object() && result.value("isError", false)) {209 return {{"error", text.empty() ? "MCP tool returned an error" : text}};210 }211 return {{"plain_text_response", text}};212}213 214json server_mcp_transport::send_rpc(const json & request, const std::function<bool()> & should_stop) {215 if (!to_server.write(request.dump())) {216 return {{"error", {{"code", -32603}, {"message", "transport closed"}}}};217 }218 219 const bool has_id = request.contains("id");220 const auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(timeout_ms);221 auto stop = [&]() {222 return (should_stop && should_stop()) || std::chrono::steady_clock::now() >= deadline;223 };224 225 std::string frame;226 while (from_server.read(frame, stop, false)) {227 json reply;228 try {229 reply = json::parse(frame);230 } catch (...) {231 if (std::chrono::steady_clock::now() >= deadline) {232 break;233 }234 continue; // skip malformed frame235 }236 // no id: a notification. mismatched id: a stale reply from a timed-out request (ids are monotonic, never a future one)237 if (!has_id || (reply.contains("id") && reply.at("id") == request.at("id"))) {238 return reply;239 }240 if (std::chrono::steady_clock::now() >= deadline) {241 break; // a flood of notifications must not outrun the deadline242 }243 }244 245 if (should_stop && should_stop()) {246 return {{"error", {{"code", -32603}, {"message", "cancelled"}}}};247 }248 if (std::chrono::steady_clock::now() >= deadline) {249 return {{"error", {{"code", -32603}, {"message", "request timed out"}}}};250 }251 return {{"error", {{"code", -32603}, {"message", "transport closed"}}}};252}253 254bool server_mcp_transport::ensure_init(const std::function<bool()> & should_stop) {255 if (initialized) {256 return true;257 }258 259 json init_req = {260 {"jsonrpc", "2.0"},261 {"id", next_id++},262 {"method", "initialize"},263 {"params", {264 {"protocolVersion", MCP_PROTOCOL_VERSION},265 {"capabilities", json::object()},266 {"clientInfo", {{"name", "llama.cpp"}, {"version", "1.0"}}},267 }},268 };269 json resp = send_rpc(init_req, should_stop);270 if (!resp.contains("result")) {271 last_error = "initialize failed: " + rpc_error_message(resp);272 return false;273 }274 275 // notifications/initialized: no id, no reply expected276 json notif = {{"jsonrpc", "2.0"}, {"method", "notifications/initialized"}};277 to_server.write(notif.dump());278 279 initialized = true;280 return true;281}282 283std::vector<server_mcp_tool_def> server_mcp_transport::list_tools(const std::function<bool()> & should_stop) {284 std::lock_guard<std::mutex> lock(rpc_mutex);285 if (!ensure_init(should_stop)) {286 return {};287 }288 if (!tools.empty()) {289 return tools;290 }291 292 json req = {{"jsonrpc", "2.0"}, {"id", next_id++}, {"method", "tools/list"}};293 json resp = send_rpc(req, should_stop);294 if (!resp.contains("result")) {295 last_error = "tools/list failed: " + rpc_error_message(resp);296 return {};297 }298 299 const json & result = resp.at("result");300 if (result.contains("tools") && result.at("tools").is_array()) {301 for (const auto & t : result.at("tools")) {302 server_mcp_tool_def def;303 def.server_name = name;304 def.name = t.value("name", "");305 def.description = t.value("description", "");306 if (t.contains("inputSchema")) {307 def.input_schema = t.at("inputSchema");308 }309 tools.push_back(std::move(def));310 }311 }312 return tools;313}314 315json server_mcp_transport::call_tool(const std::string & tool_name,316 const json & arguments,317 const std::function<bool()> & should_stop) {318 std::lock_guard<std::mutex> lock(rpc_mutex);319 if (!ensure_init(should_stop)) {320 return {{"error", last_error}};321 }322 323 json req = {324 {"jsonrpc", "2.0"},325 {"id", next_id++},326 {"method", "tools/call"},327 {"params", {{"name", tool_name}, {"arguments", arguments}}},328 };329 json resp = send_rpc(req, should_stop);330 if (resp.contains("error")) {331 return {{"error", rpc_error_message(resp)}};332 }333 if (resp.contains("result")) {334 return mcp_result_to_response(resp.at("result"));335 }336 return {{"error", "invalid response from MCP server"}};337}338 339//340// server_mcp_stdio341//342 343struct server_mcp_stdio::process_handle {344 common_subproc sp;345 FILE * in = nullptr; // child stdin346 FILE * out = nullptr; // child stdout347 FILE * err = nullptr; // child stderr348};349 350#if defined(_WIN32)351// 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 page352static std::wstring windows_utf8_to_wide(const std::string & s) {353 if (s.empty()) {354 return std::wstring();355 }356 int n = MultiByteToWideChar(CP_UTF8, 0, s.data(), (int) s.size(), NULL, 0);357 if (n <= 0) {358 return std::wstring();359 }360 std::wstring w((size_t) n, L'\0');361 MultiByteToWideChar(CP_UTF8, 0, s.data(), (int) s.size(), &w[0], n);362 return w;363}364 365static std::string windows_wide_to_utf8(const wchar_t * s, int len /* -1 for NUL-terminated */) {366 int n = WideCharToMultiByte(CP_UTF8, 0, s, len, NULL, 0, NULL, NULL);367 if (n <= 0) {368 return std::string();369 }370 std::string out((size_t) n, '\0');371 WideCharToMultiByte(CP_UTF8, 0, s, len, &out[0], n, NULL, NULL);372 if (len == -1 && !out.empty() && out.back() == '\0') {373 out.pop_back(); // drop the terminator WideCharToMultiByte counts for -1374 }375 return out;376}377#endif378 379static std::string mcp_resolve_command(const std::string & command) {380#if defined(_WIN32)381 // For Windows: make sure we handle ".exe" correctly, as well as UTF-8382 std::wstring wcmd = windows_utf8_to_wide(command);383 wchar_t buf[MAX_PATH * 4];384 const DWORD cap = (DWORD) (sizeof(buf) / sizeof(buf[0]));385 386 auto search = [&](const wchar_t * ext) -> std::string {387 DWORD n = SearchPathW(NULL, wcmd.c_str(), ext, cap, buf, NULL);388 return (n > 0 && n < cap) ? windows_wide_to_utf8(buf, (int) n) : std::string();389 };390 391 std::string found = search(NULL); // exact path / already-extensioned / .exe on PATH392 if (!found.empty()) {393 return found;394 }395 396 std::wstring pathext;397 DWORD need = GetEnvironmentVariableW(L"PATHEXT", NULL, 0);398 if (need > 0) {399 pathext.resize(need);400 DWORD got = GetEnvironmentVariableW(L"PATHEXT", &pathext[0], need);401 pathext.resize(got);402 }403 if (pathext.empty()) {404 pathext = L".COM;.EXE;.BAT;.CMD";405 }406 for (size_t start = 0; start <= pathext.size();) {407 size_t sep = pathext.find(L';', start);408 std::wstring ext = pathext.substr(start, sep == std::wstring::npos ? std::wstring::npos : sep - start);409 if (!ext.empty()) {410 found = search(ext.c_str());411 if (!found.empty()) {412 return found;413 }414 }415 if (sep == std::wstring::npos) {416 break;417 }418 start = sep + 1;419 }420 return command; // give up and let subprocess.h report the spawn error421#else422 return command;423#endif // _WIN32424}425 426static std::vector<std::string> mcp_parent_env() {427 std::vector<std::string> env;428#if defined(_WIN32)429 LPWCH block = GetEnvironmentStringsW();430 if (block) {431 for (LPWCH e = block; *e; e += wcslen(e) + 1) {432 env.emplace_back(windows_wide_to_utf8(e, -1));433 }434 FreeEnvironmentStringsW(block);435 }436#else437 if (environ) {438 for (char ** e = environ; *e; ++e) {439 env.emplace_back(*e);440 }441 }442#endif443 return env;444}445 446// parent env with the config overrides applied, in "KEY=VALUE" form447static std::vector<std::string> mcp_build_env(const std::map<std::string, std::string> & overrides) {448 std::vector<std::string> env;449 for (auto & e : mcp_parent_env()) {450 size_t eq = e.find('=');451 std::string key = eq == std::string::npos ? e : e.substr(0, eq);452 if (overrides.find(key) == overrides.end()) {453 env.push_back(e);454 }455 }456 for (auto & [k, v] : overrides) {457 env.push_back(k + "=" + v);458 }459 return env;460}461 462server_mcp_stdio::server_mcp_stdio(const server_mcp_server_config & config) : config(config) {463 name = config.name;464 timeout_ms = config.timeout_ms;465 // bound the reply queue: send_rpc only drains during a call, so unsolicited notifications would otherwise grow it without limit466 from_server.max_size = 65536;467}468 469server_mcp_stdio::~server_mcp_stdio() {470 join_pumps();471}472 473bool server_mcp_stdio::start() {474 std::vector<std::string> argv_s;475 argv_s.push_back(mcp_resolve_command(config.command));476 argv_s.insert(argv_s.end(), config.args.begin(), config.args.end());477 478 int options = subprocess_option_no_window | subprocess_option_search_user_path;479 std::vector<std::string> envp_s;480 if (config.env.empty()) {481 options |= subprocess_option_inherit_environment;482 } else {483 envp_s = mcp_build_env(config.env);484 }485 486 auto handle = std::make_unique<process_handle>();487 bool ok = handle->sp.create(argv_s, options, envp_s, config.cwd.empty() ? nullptr : config.cwd.c_str());488 if (!ok) {489 SRV_WRN("MCP '%s': failed to spawn '%s'\n", config.name.c_str(), config.command.c_str());490 return false;491 }492 handle->in = handle->sp.stdin_file();493 handle->out = handle->sp.stdout_file();494 handle->err = handle->sp.stderr_file();495 496 proc = std::move(handle);497 running.store(true);498 reader = std::thread([this] { reader_loop(); });499 writer = std::thread([this] { writer_loop(); });500 errlog = std::thread([this] { errlog_loop(); });501 return true;502}503 504void server_mcp_stdio::close() {505 join_pumps();506}507 508bool server_mcp_stdio::is_alive() const {509 return running.load();510}511 512std::string server_mcp_stdio::diagnostics() {513 std::string out;514 {515 std::lock_guard<std::mutex> lock(rpc_mutex); // last_error is written by send_rpc's callers516 out = last_error;517 }518 std::lock_guard<std::mutex> lk(err_mu);519 if (!err_tail.empty()) {520 if (!out.empty()) {521 out += "; ";522 }523 out += "last stderr: " + err_tail;524 }525 return out;526}527 528void server_mcp_stdio::reader_loop() {529 mcp_pump_ndjson(proc->out, running, [this](std::string && line) {530 return from_server.write(std::move(line)); // false => consumer gone, stop531 });532 running.store(false);533 to_server.close_write(); // stop the writer534 from_server.close_write(); // EOF to any waiting caller535}536 537// 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.538static bool mcp_write_all(FILE * f, const std::string & data, std::atomic<bool> & running) {539 if (!f) {540 return false;541 }542 size_t total = 0;543#if defined(_WIN32)544 HANDLE h = (HANDLE) _get_osfhandle(_fileno(f));545 DWORD nowait = PIPE_NOWAIT;546 SetNamedPipeHandleState(h, &nowait, NULL, NULL);547 while (total < data.size() && running.load()) {548 DWORD written = 0;549 BOOL ok = WriteFile(h, data.data() + total, (DWORD) (data.size() - total), &written, NULL);550 if (ok && written > 0) {551 total += written;552 continue;553 }554 if (!ok) {555 DWORD err = GetLastError();556 if (err != ERROR_NO_DATA && err != ERROR_PIPE_BUSY) {557 return false;558 }559 }560 // backpressure (pipe full) is rare for small JSON-RPC frames; sleep rather than spin.561 // no writable-wait exists for a PIPE_NOWAIT anonymous pipe, so this polls like the POSIX poll() path.562 std::this_thread::sleep_for(std::chrono::milliseconds(10));563 }564#else565 int fd = fileno(f);566 int fl = fcntl(fd, F_GETFL, 0);567 if (fl >= 0) {568 fcntl(fd, F_SETFL, fl | O_NONBLOCK);569 }570 while (total < data.size() && running.load()) {571 ssize_t n = write(fd, data.data() + total, data.size() - total);572 if (n > 0) {573 total += (size_t) n;574 continue;575 }576 if (n == 0) {577 return false;578 }579 if (errno == EINTR) {580 continue;581 }582 if (errno != EAGAIN && errno != EWOULDBLOCK) {583 return false;584 }585 struct pollfd pfd;586 pfd.fd = fd;587 pfd.events = POLLOUT;588 pfd.revents = 0;589 int pr = poll(&pfd, 1, 50);590 if (pr < 0) {591 if (errno == EINTR) {592 continue;593 }594 return false;595 }596 if (pfd.revents & (POLLERR | POLLNVAL | POLLHUP)) {597 return false;598 }599 }600#endif601 return total == data.size();602}603 604void server_mcp_stdio::writer_loop() {605 auto should_stop = [this] { return !running.load(); };606 std::string msg;607 while (to_server.read(msg, should_stop)) {608 msg.push_back('\n');609 if (!mcp_write_all(proc->in, msg, running)) {610 break; // child gone or shutting down611 }612 }613 running.store(false);614 to_server.close_read(); // fail fast on any further send_rpc write615 from_server.close_write(); // wake any caller waiting for a reply616}617 618void server_mcp_stdio::errlog_loop() {619 static constexpr size_t ERR_TAIL_MAX = 4096;620 // drain stderr (an undrained pipe blocks the child):621 // log it, and keep a bounded tail for reporting when the server dies622 mcp_pump_ndjson(proc->err, running, [this](std::string && line) {623 SRV_DBG("MCP '%s' stderr: %s\n", name.c_str(), line.c_str());624 std::lock_guard<std::mutex> lk(err_mu);625 err_tail += line;626 err_tail += '\n';627 if (err_tail.size() > ERR_TAIL_MAX) {628 err_tail.erase(0, err_tail.size() - ERR_TAIL_MAX);629 }630 return true;631 });632}633 634void server_mcp_stdio::join_pumps() {635 if (!proc) {636 return;637 }638 running.store(false);639 to_server.close_write(); // wake the writer if it waits for a message640 from_server.close_write(); // wake any caller waiting for a reply641 642 proc->sp.terminate(); // child death unblocks the blocked fread/fwrite643 644 if (writer.joinable()) writer.join();645 if (reader.joinable()) reader.join();646 if (errlog.joinable()) errlog.join();647 648 proc->sp.join(); // reap the child: never waiting would leave the pid a zombie for the process lifetime649 proc.reset();650}651 652 653//654// server_mcp655//656 657static constexpr int MCP_COOLDOWN_SECONDS = 5;658static constexpr int MCP_WARMUP_TIMEOUT_SECONDS = 10; // cap per-server tool discovery at startup659 660server_mcp::~server_mcp() {661 shutdown();662 663 std::vector<std::shared_ptr<server_mcp_transport>> to_close;664 {665 std::lock_guard<std::mutex> lock(mutex);666 for (auto & [name, t] : transports) {667 to_close.push_back(std::move(t));668 }669 transports.clear();670 }671 for (auto & t : to_close) {672 t->close();673 }674}675 676std::shared_ptr<server_mcp_transport> server_mcp::create_transport(const server_mcp_server_config & cfg) {677 return std::make_shared<server_mcp_stdio>(cfg);678}679 680void server_mcp::shutdown() {681 stopping.store(true);682}683 684const server_mcp_server_config * server_mcp::find_config(const std::string & name) const {685 for (const auto & c : configs) {686 if (c.name == name) {687 return &c;688 }689 }690 return nullptr;691}692 693void server_mcp::start(const common_params & params) {694 auto append = [this](const std::string & json_str) {695 try {696 auto parsed = server_mcp_server_config::parse_from_json(json_str);697 if (parsed.empty()) {698 SRV_WRN("%s", "MCP config: no servers found in JSON\n");699 }700 for (auto & p : parsed) {701 // names must be unique across both config sources: get_or_create / find_config key on the name702 if (find_config(p.name)) {703 SRV_WRN("MCP config: duplicate server name '%s', skipping\n", p.name.c_str());704 continue;705 }706 configs.push_back(std::move(p));707 }708 } catch (const std::exception & e) {709 throw std::runtime_error(std::string("failed to parse MCP config JSON: ") + e.what());710 }711 };712 if (!params.mcp_servers_config.empty()) {713 std::ifstream f = fs_open_ifstream(params.mcp_servers_config, std::ios::in);714 if (!f) {715 throw std::runtime_error("failed to open MCP config file: " + params.mcp_servers_config);716 }717 std::stringstream ss;718 ss << f.rdbuf();719 append(ss.str());720 }721 if (!params.mcp_servers_json.empty()) {722 append(params.mcp_servers_json);723 }724 725 if (configs.empty()) {726 return;727 }728 729 std::vector<server_mcp_tool_def> discovered;730 for (const auto & cfg : configs) {731 auto t = create_transport(cfg);732 if (!t->start()) {733 SRV_WRN("MCP warmup: failed to spawn '%s': %s\n", cfg.name.c_str(), t->diagnostics().c_str());734 continue;735 }736 // bound warmup per server so an unresponsive one can't stall startup for the full per-call timeout737 const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(MCP_WARMUP_TIMEOUT_SECONDS);738 auto should_stop = [this, deadline]() {739 return stopping.load() || std::chrono::steady_clock::now() >= deadline;740 };741 auto tools = t->list_tools(should_stop);742 SRV_INF("MCP warmup: '%s' discovered %zu tools\n", cfg.name.c_str(), tools.size());743 discovered.insert(discovered.end(), tools.begin(), tools.end());744 t->close();745 }746 747 std::lock_guard<std::mutex> lock(mutex);748 registry.swap(discovered);749}750 751std::vector<server_mcp_tool_def> server_mcp::list_tools() const {752 std::lock_guard<std::mutex> lock(mutex);753 return registry;754}755 756json server_mcp::call_tool(const std::string & server_name,757 const std::string & tool_name,758 const json & arguments,759 const std::function<bool()> & should_stop) {760 auto transport = get_or_create(server_name);761 if (!transport) {762 return {{"error", "MCP server unavailable: " + server_name}};763 }764 765 auto stop = [this, &should_stop]() {766 return stopping.load() || (should_stop && should_stop());767 };768 return transport->call_tool(tool_name, arguments, stop);769}770 771std::shared_ptr<server_mcp_transport> server_mcp::get_or_create(const std::string & name) {772 std::vector<std::shared_ptr<server_mcp_transport>> to_close; // closed after unlock773 std::shared_ptr<server_mcp_transport> result;774 775 {776 std::lock_guard<std::mutex> lock(mutex);777 if (stopping.load()) {778 return nullptr;779 }780 781 auto now = std::chrono::steady_clock::now();782 auto dead_it = dead_servers.find(name);783 if (dead_it != dead_servers.end()) {784 if (now < dead_it->second) {785 return nullptr;786 }787 dead_servers.erase(dead_it);788 }789 790 auto it = transports.find(name);791 if (it != transports.end()) {792 if (it->second->is_alive()) {793 return it->second;794 }795 SRV_WRN("MCP '%s' is no longer alive: %s\n", name.c_str(), it->second->diagnostics().c_str());796 to_close.push_back(std::move(it->second));797 transports.erase(it);798 }799 800 const server_mcp_server_config * cfg = find_config(name);801 if (cfg) {802 auto fresh = create_transport(*cfg);803 if (fresh->start() && fresh->is_alive()) {804 transports[name] = fresh;805 result = fresh;806 } else {807 SRV_WRN("MCP '%s': failed to start: %s\n", name.c_str(), fresh->diagnostics().c_str());808 to_close.push_back(std::move(fresh));809 dead_servers[name] = now + std::chrono::seconds(MCP_COOLDOWN_SECONDS);810 }811 }812 }813 814 for (auto & t : to_close) {815 t->close(); // blocking call, no leaks816 }817 818 return result;819}820 821