Commit d8c8cc4d9 for llama.cpp
commit d8c8cc4d9b225e4a3bc552a8ecec949cd255bb39
Author: Hamish M. Blair <hmblair@stanford.edu>
Date: Fri Nov 13 03:19:26 2026 +1100
server : cancel MCP tool calls on client disconnect or timeout (#32025)
* server : cancel MCP tool calls on client disconnect or timeout
* server : shorten MCP cancellation tests and fix a type-check error
diff --git a/tools/server/README-dev.md b/tools/server/README-dev.md
index 01d1a65cf..85fb3f78e 100644
--- a/tools/server/README-dev.md
+++ b/tools/server/README-dev.md
@@ -239,6 +239,8 @@ That requires `JSON.stringify` when formatted to message content:
}
```
+When the client disconnects during an MCP tool call, or the call exceeds `timeout_ms`, `llama-server` sends `notifications/cancelled` with the request id to the MCP server and returns an error.
+
Set `stream: true` in the request body to stream a tool's output as it runs, instead of waiting for it to finish. Only certain tools accept this (for ex. `exec_shell_command`);
returns 404 if tool doesn't support it.
diff --git a/tools/server/server-mcp.cpp b/tools/server/server-mcp.cpp
index 22401e08a..08929ba54 100644
--- a/tools/server/server-mcp.cpp
+++ b/tools/server/server-mcp.cpp
@@ -244,14 +244,30 @@ json server_mcp_transport::send_rpc(const json & request, const std::function<bo
}
if (should_stop && should_stop()) {
+ notify_cancelled(request, "cancelled");
return {{"error", {{"code", -32603}, {"message", "cancelled"}}}};
}
if (std::chrono::steady_clock::now() >= deadline) {
+ notify_cancelled(request, "request timed out");
return {{"error", {{"code", -32603}, {"message", "request timed out"}}}};
}
return {{"error", {{"code", -32603}, {"message", "transport closed"}}}};
}
+// MCP cancellation: a notification that lets the server abandon work nobody waits for anymore
+// (e.g. a slow tool call after the client dropped the connection). the spec forbids cancelling initialize.
+void server_mcp_transport::notify_cancelled(const json & request, const std::string & reason) {
+ if (!request.contains("id") || request.value("method", "") == "initialize") {
+ return;
+ }
+ json notif = {
+ {"jsonrpc", "2.0"},
+ {"method", "notifications/cancelled"},
+ {"params", {{"requestId", request.at("id")}, {"reason", reason}}},
+ };
+ to_server.write(notif.dump());
+}
+
bool server_mcp_transport::ensure_init(const std::function<bool()> & should_stop) {
if (initialized) {
return true;
diff --git a/tools/server/server-mcp.h b/tools/server/server-mcp.h
index c7f33a3f7..2b4eb8920 100644
--- a/tools/server/server-mcp.h
+++ b/tools/server/server-mcp.h
@@ -82,9 +82,10 @@ protected:
std::string last_error;
std::vector<server_mcp_tool_def> tools;
- // both assume rpc_mutex is already held by the public caller
+ // all 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": ...}
+ void notify_cancelled(const json & request, const std::string & reason); // notifications/cancelled for an abandoned request
};
//
diff --git a/tools/server/server-tools.cpp b/tools/server/server-tools.cpp
index 8070e14b4..c92f1e9f8 100644
--- a/tools/server/server-tools.cpp
+++ b/tools/server/server-tools.cpp
@@ -896,7 +896,7 @@ struct server_tool_read_file : server_tool {
};
}
- json invoke(json params, server_tool::stream *) const override {
+ json invoke(json params, server_tool::stream *, const std::function<bool()> &) const override {
std::string path = params.at("path").get<std::string>();
int start_line = json_value(params, "start_line", 1);
int end_line = json_value(params, "end_line", -1); // -1 = no limit
@@ -1013,7 +1013,7 @@ struct server_tool_file_glob_search : server_tool {
};
}
- json invoke(json params, server_tool::stream *) const override {
+ json invoke(json params, server_tool::stream *, const std::function<bool()> &) const override {
auto io = make_tools_io(params);
const std::string path = params.at("path").get<std::string>();
@@ -1128,7 +1128,7 @@ struct server_tool_grep_search : server_tool {
};
}
- json invoke(json params, server_tool::stream *) const override {
+ json invoke(json params, server_tool::stream *, const std::function<bool()> &) const override {
std::string path = params.at("path").get<std::string>();
std::string pat_str = params.at("pattern").get<std::string>();
std::string include = json_value(params, "include", std::string("**"));
@@ -1271,7 +1271,7 @@ struct server_tool_exec_shell_command : server_tool {
};
}
- json invoke(json params, server_tool::stream * st) const override {
+ json invoke(json params, server_tool::stream * st, const std::function<bool()> &) const override {
std::string command = params.at("command").get<std::string>();
int timeout = json_value(params, "timeout", 10);
size_t max_output = (size_t) json_value(params, "max_output_size", (int) SERVER_TOOL_EXEC_SHELL_COMMAND_MAX_OUTPUT_SIZE);
@@ -1348,7 +1348,7 @@ struct server_tool_write_file : server_tool {
};
}
- json invoke(json params, server_tool::stream *) const override {
+ json invoke(json params, server_tool::stream *, const std::function<bool()> &) const override {
std::string path = params.at("path").get<std::string>();
std::string content = params.at("content").get<std::string>();
@@ -1405,7 +1405,7 @@ struct server_tool_edit_file : server_tool {
};
}
- json invoke(json params, server_tool::stream *) const override {
+ json invoke(json params, server_tool::stream *, const std::function<bool()> &) const override {
std::string path = params.at("path").get<std::string>();
const json & edits_json = params.at("edits");
@@ -1721,7 +1721,7 @@ struct server_tool_get_info : server_tool {
};
}
- json invoke(json params, server_tool::stream *) const override {
+ json invoke(json params, server_tool::stream *, const std::function<bool()> &) const override {
auto io = make_tools_io(params);
// inside an isolate, we always use the linux command
@@ -1835,8 +1835,8 @@ struct server_mcp_tool : server_tool {
};
}
- json invoke(json params, server_tool::stream *) const override {
- return mcp_mgr.call_tool(server_name, tool_name, params);
+ json invoke(json params, server_tool::stream *, const std::function<bool()> & should_stop) const override {
+ return mcp_mgr.call_tool(server_name, tool_name, params, should_stop);
}
};
@@ -2133,7 +2133,7 @@ void server_tools::setup(const std::vector<std::string> & enabled_tools,
auto done = std::make_unique<server_tool_stream_result>();
try {
- tool.invoke(params, &st);
+ tool.invoke(params, &st, req.should_stop);
} catch (const std::exception & e) {
done->error_msg = e.what();
} catch (...) {
@@ -2158,7 +2158,7 @@ void server_tools::setup(const std::vector<std::string> & enabled_tools,
return true;
};
} else {
- json result = tool.invoke(params, nullptr);
+ json result = tool.invoke(params, nullptr, req.should_stop);
res->status = 200;
res->data = safe_json_to_str(result);
}
diff --git a/tools/server/server-tools.h b/tools/server/server-tools.h
index e7332f2e5..5a63ee973 100644
--- a/tools/server/server-tools.h
+++ b/tools/server/server-tools.h
@@ -26,7 +26,8 @@ struct server_tool {
std::function<bool()> alive;
void push(const std::string & chunk);
};
- virtual json invoke(json params, stream * st = nullptr) const = 0;
+ // should_stop: true once the caller stopped waiting, e.g. the client dropped the connection; a null should_stop means "never stop"
+ virtual json invoke(json params, stream * st = nullptr, const std::function<bool()> & should_stop = nullptr) const = 0;
json to_json() const;
};
diff --git a/tools/server/tests/fixtures/mcp_slow_server.py b/tools/server/tests/fixtures/mcp_slow_server.py
index 7f8e67835..879d48ffe 100644
--- a/tools/server/tests/fixtures/mcp_slow_server.py
+++ b/tools/server/tests/fixtures/mcp_slow_server.py
@@ -1,6 +1,7 @@
#!/usr/bin/env python3
"""
MCP server that sleeps before responding, for timeout testing.
+With --log, it appends every message it reads from stdin to that file, for cancellation testing.
"""
import json
import sys
@@ -70,6 +71,7 @@ HANDLERS = {
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--delay", type=float, default=5.0, help="Delay in seconds for sleep tool")
+ parser.add_argument("--log", help="File to append every message read from stdin to")
args = parser.parse_args()
# Override the sleep duration
@@ -102,6 +104,9 @@ def main():
line = line.strip()
if not line:
continue
+ if args.log:
+ with open(args.log, "a") as log:
+ log.write(line + "\n")
try:
request = json.loads(line)
except json.JSONDecodeError:
diff --git a/tools/server/tests/unit/test_mcp_servers.py b/tools/server/tests/unit/test_mcp_servers.py
index 877b732a5..7a9f0bea5 100644
--- a/tools/server/tests/unit/test_mcp_servers.py
+++ b/tools/server/tests/unit/test_mcp_servers.py
@@ -9,6 +9,7 @@ Invariants verified:
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
+7. The MCP server receives notifications/cancelled when the client drops or the call times out
"""
import json
import os
@@ -319,29 +320,60 @@ def test_mcp_concurrent_tool_calls():
server.stop()
-def test_mcp_tool_timeout():
- """Tool call should timeout if MCP server is too slow."""
- global server
+def _slow_server_config(log_path: str, timeout_ms: int) -> str:
MCP_SLOW_SERVER = os.path.join(FIXTURES_DIR, "mcp_slow_server.py")
- mcp_json = _mcp_config_json({
+ return _mcp_config_json({
"slow": {
"command": sys.executable,
- "args": [MCP_SLOW_SERVER, "--delay", "5"],
- "timeout_ms": 500
+ "args": [MCP_SLOW_SERVER, "--log", log_path],
+ "timeout_ms": timeout_ms,
}
})
- server = _start_server_with_mcp(mcp_json)
+
+
+def _messages_read_by_mcp_server(log_path: str) -> list:
+ """Parses the fixture's log, skipping a last line that the fixture is still writing."""
+ with open(log_path) as f:
+ return [json.loads(line) for line in f if line.endswith("\n")]
+
+
+def _wait_for_cancellation(log_path: str, limit_seconds: float) -> dict:
+ """Returns the notifications/cancelled message once the MCP server has read it."""
+ deadline = time.time() + limit_seconds
+ while time.time() < deadline:
+ for message in _messages_read_by_mcp_server(log_path):
+ if message.get("method") == "notifications/cancelled":
+ return message
+ time.sleep(0.1)
+ raise AssertionError(f"the MCP server read no notifications/cancelled: {_messages_read_by_mcp_server(log_path)}")
+
+
+def _tools_call_id(log_path: str):
+ """Returns the id of the one tools/call request the MCP server has read."""
+ ids = [m["id"] for m in _messages_read_by_mcp_server(log_path) if m.get("method") == "tools/call"]
+ assert len(ids) == 1, ids
+ return ids[0]
+
+
+def test_mcp_tool_timeout():
+ """A tool call that exceeds timeout_ms returns an error and sends notifications/cancelled to the MCP server."""
+ global server
+ with tempfile.NamedTemporaryFile(suffix=".jsonl", delete=False) as f:
+ log_path = f.name
+ server = _start_server_with_mcp(_slow_server_config(log_path, timeout_ms=500))
try:
res = server.make_request("POST", "/tools", data={
"tool": "slow_sleep",
- "params": {"seconds": 5}
+ "params": {"seconds": 1}
})
assert res.status_code == 200, res.body
- body = res.body
- assert "error" in body, body
+ assert "error" in res.body, res.body
+ cancellation = _wait_for_cancellation(log_path, limit_seconds=10)
+ assert cancellation["params"]["requestId"] == _tools_call_id(log_path)
finally:
server.stop()
+ os.unlink(log_path)
def test_mcp_warmup_partial_failure():
@@ -711,3 +743,23 @@ def test_mcp_grandchild_holding_pipes_does_not_deadlock():
assert elapsed < 3, f"server shutdown took {elapsed:.1f}s (expected < 3s) — teardown likely deadlocked"
finally:
server.stop()
+
+
+def test_mcp_dropped_client_sends_cancellation():
+ """Dropping the connection during a tool call sends notifications/cancelled for that request to the MCP server."""
+ global server
+ with tempfile.NamedTemporaryFile(suffix=".jsonl", delete=False) as f:
+ log_path = f.name
+ server = _start_server_with_mcp(_slow_server_config(log_path, timeout_ms=30000))
+
+ try:
+ with pytest.raises(requests.exceptions.ReadTimeout):
+ server.make_request("POST", "/tools", data={
+ "tool": "slow_sleep",
+ "params": {"seconds": 1.5}
+ }, timeout=0.5)
+ cancellation = _wait_for_cancellation(log_path, limit_seconds=10)
+ assert cancellation["params"]["requestId"] == _tools_call_id(log_path)
+ finally:
+ server.stop()
+ os.unlink(log_path)