From 774bd66570d5299620e6e6b619cb5b094d429ac0 Mon Sep 17 00:00:00 2001 From: Sander Striker Date: Sat, 31 Oct 2026 14:14:44 +0100 Subject: [PATCH 1/5] feat: speak MCP 2026-07-28 in the hand-written MCP code MCP 2026-07-28 removes the initialize handshake and the session. Each request names its protocol version and client capabilities in _meta. Each streamable HTTP request mirrors its version, method and name in headers. Each result has a resultType. rmcp 3.5 already does this for the memory server. This change does it for the three parts of amem that do not use rmcp. amem tool-server now serves both revisions. A request that names its version in _meta, or that asks server/discover, gets 2026-07-28 behaviour. The server checks the version, the client capabilities, and the MCP-Protocol-Version, Mcp-Method and Mcp-Name headers. It refuses a missing header with -32020, as the spec requires. It refuses an unknown version with -32022 and lists the versions that it supports. Results get resultType "complete" and the server info. An initialize request still opens the old path. initialize now answers with the version that the client asks for if the server supports it, and with 2025-11-25 if not. Before, it repeated any version. The hook client in the sandbox now sends one stateless tools/call. Before, it sent three requests. A test runs it against the real memory server through the egress router. The hosted-tool client sends server/discover first. If the answer lists 2026-07-28, it sends one stateless tools/call. If not, it uses the initialize handshake. On that path it now also sends the negotiated MCP-Protocol-Version header. It keeps the result of the probe per server URL and forgets it when a call fails. Both clients refuse a resultType other than "complete" and accept a result that has no resultType. The new module mcp_meta holds the shared parts: the _meta keys, the headers, the base64 encoding of header values, and the resultType check. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Bhj4Ck4y5iwZUsjpgQx8A2 --- Cargo.lock | 1 + crates/memory/Cargo.toml | 1 + crates/memory/src/hosted.rs | 338 ++++++++++++++++----- crates/memory/src/main.rs | 1 + crates/memory/src/mcp_http.rs | 85 +++--- crates/memory/src/mcp_meta.rs | 203 +++++++++++++ crates/memory/src/tool_server.rs | 504 ++++++++++++++++++++++++++++--- crates/memory/tests/hosted.rs | 111 ++++++- 8 files changed, 1089 insertions(+), 155 deletions(-) create mode 100644 crates/memory/src/mcp_meta.rs diff --git a/Cargo.lock b/Cargo.lock index aea7c810..6143a87e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -43,6 +43,7 @@ name = "amem" version = "0.1.0" dependencies = [ "axum", + "base64 0.22.1", "clap", "dm-client", "dm-core", diff --git a/crates/memory/Cargo.toml b/crates/memory/Cargo.toml index 33cbae4a..73f4e19d 100644 --- a/crates/memory/Cargo.toml +++ b/crates/memory/Cargo.toml @@ -7,6 +7,7 @@ repository.workspace = true [dependencies] axum = { version = "0.8", default-features = false, features = ["http1", "tokio", "json"] } +base64 = "0.22" clap = { version = "4.6.7", features = ["derive", "env"] } dm-client = { version = "0.1.0", path = "../client" } dm-core = { version = "0.1.0", path = "../core" } diff --git a/crates/memory/src/hosted.rs b/crates/memory/src/hosted.rs index a6201a80..dde64625 100644 --- a/crates/memory/src/hosted.rs +++ b/crates/memory/src/hosted.rs @@ -8,19 +8,24 @@ //! names one the policy refuses, is not kept; what is kept says where it was fetched //! from. //! +//! The server is asked once what MCP it speaks (`server/discover`): one of 2026-07-28 +//! is called with one stateless request, one before it, which has no such method, +//! after the `initialize` handshake, as every server was before that revision. +//! //! The call leaves through the egress ([`Egress::post`]), as a model call does: the //! session's label is held to the tool's ceiling before anything is sent, the server's //! bearer goes on there, and every request is metered and traced with its status and //! latency. The harness itself cannot reach the server that way; a tool server is a //! destination of the launcher's, not of the sandbox's. +use crate::mcp_meta::{self, supported, LEGACY, MODERN, MODERN_ERRORS, UNSUPPORTED_VERSION}; use dm_egress::{Auth, Destination, Egress}; use dm_policy::Grants; use dm_server::sources::Answered; use dm_server::Memory; use serde_json::{json, Value}; use std::collections::HashMap; -use std::sync::Arc; +use std::sync::{Arc, Mutex, OnceLock}; /// What a tool server says back for one call: the tool's JSON, and where it went. #[derive(Debug, PartialEq, Eq)] @@ -110,6 +115,7 @@ pub fn reply(message: &Value) -> Result { return Err(format!("the tool server refused the call: {e}")); } let result = &message["result"]; + crate::mcp_meta::complete(result).map_err(|e| format!("the tool server: {e}"))?; let text = result["content"] .as_array() .and_then(|c| c.iter().find_map(|b| b["text"].as_str())) @@ -159,7 +165,207 @@ pub fn checked(memory: &Memory, tool: &str, r: Reply) -> Result &'static Mutex> { + static ERAS: OnceLock>> = OnceLock::new(); + ERAS.get_or_init(Mutex::default) +} + +fn known(url: &str) -> Option { + eras() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .get(url) + .copied() +} + +fn learn(url: &str, era: Option) { + let mut eras = eras() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + match era { + Some(e) => eras.insert(url.to_string(), e), + None => eras.remove(url), + }; +} + +/// What a tool server's answer to `server/discover` says it speaks. A result listing +/// 2026-07-28 is that revision; one listing only revisions with a handshake, or that +/// revision's refusal of the version naming only those, is one of them. Anything else +/// no server of 2026-07-28 answers, so it comes from one before it that has no such +/// method: an error status without one of that revision's errors, a method unknown, +/// nothing at all. +/// +/// # Errors +/// +/// That revision's refusal of the request itself, or a list of versions none of +/// which this launcher speaks. +fn era(content_type: &str, body: &str) -> Result { + let Ok(m) = message(content_type, body) else { + return Ok(Era::Handshake); + }; + let listed = if let Some(e) = m.get("error") { + let code = e["code"].as_i64().unwrap_or_default(); + if !MODERN_ERRORS.contains(&code) { + return Ok(Era::Handshake); + } + if code != UNSUPPORTED_VERSION { + return Err(format!("the tool server refused server/discover: {e}")); + } + &e["data"]["supported"] + } else { + &m["result"]["supportedVersions"] + }; + let Some(listed) = listed.as_array() else { + return Ok(Era::Handshake); + }; + let listed: Vec<&str> = listed.iter().filter_map(Value::as_str).collect(); + if listed.contains(&MODERN) && m.get("error").is_none() { + Ok(Era::Stateless) + } else if listed.iter().any(|v| LEGACY.contains(v)) { + Ok(Era::Handshake) + } else { + Err(format!( + "the tool server speaks MCP {}, and this launcher {}", + listed.join(", "), + supported().join(", ") + )) + } +} + +/// One POST to a tool server, as it answered. +struct Sent { + status: u16, + content_type: String, + session: Option, + body: String, +} + +impl Sent { + /// The answer, when its status is a success. + fn ok(self) -> Result { + if (200..300).contains(&self.status) { + Ok(self) + } else { + Err(format!( + "the tool server answered {}: {}", + self.status, + self.body.trim() + )) + } + } +} + +/// The way to one tool server: through the egress, under the tool's name, to the path +/// of its MCP endpoint. +struct Server<'a> { + egress: &'a Egress, + tool: &'a str, + path: String, +} + +impl Server<'_> { + async fn post(&self, body: &Value, extra: &[(&'static str, String)]) -> Result { + let mut headers = axum::http::HeaderMap::new(); + headers.insert( + axum::http::header::CONTENT_TYPE, + axum::http::HeaderValue::from_static("application/json"), + ); + headers.insert( + axum::http::header::ACCEPT, + axum::http::HeaderValue::from_static("application/json, text/event-stream"), + ); + for (name, value) in extra { + let value = axum::http::HeaderValue::from_str(value) + .map_err(|_| format!("{name}: {value:?} is not a header value"))?; + headers.insert(axum::http::HeaderName::from_static(name), value); + } + let r = self + .egress + .post(self.tool, &self.path, headers, body.to_string().into()) + .await + .map_err(|e| e.to_string())?; + let text = |name: &str| { + r.headers + .get(name) + .and_then(|v| v.to_str().ok()) + .map(str::to_string) + }; + Ok(Sent { + status: r.status, + content_type: text("content-type").unwrap_or_default(), + session: text("mcp-session-id"), + body: r.body, + }) + } + + /// Ask the server what it speaks. + async fn discover(&self) -> Result { + let request = mcp_meta::request(1, "server/discover", Value::Null, "amem"); + let sent = self.post(&request, &mcp_meta::headers(&request)).await?; + era(&sent.content_type, &sent.body) + } + + /// One `tools/call` of 2026-07-28: one request, no session. + async fn stateless(&self, params: Value) -> Result { + let request = mcp_meta::request(1, "tools/call", params, "amem"); + self.post(&request, &mcp_meta::headers(&request)) + .await? + .ok() + } + + /// One `tools/call` after the handshake of the revisions before 2026-07-28, in the + /// session the server opens, if it opens one, at the version it answers with. + async fn handshake(&self, params: Value) -> Result { + let opened = self + .post( + &json!({ + "jsonrpc": "2.0", "id": 1, "method": "initialize", + "params": {"protocolVersion": "2025-06-18", "capabilities": {}, + "clientInfo": {"name": "amem", "version": env!("CARGO_PKG_VERSION")}} + }), + &[], + ) + .await? + .ok()?; + let mut headers = Vec::new(); + if let Some(s) = opened.session.clone() { + headers.push(("mcp-session-id", s)); + } + let version = message(&opened.content_type, &opened.body) + .ok() + .and_then(|m| m["result"]["protocolVersion"].as_str().map(str::to_string)); + if let Some(v) = version { + headers.push((mcp_meta::VERSION_HEADER, v)); + } + self.post( + &json!({"jsonrpc": "2.0", "method": "notifications/initialized"}), + &headers, + ) + .await? + .ok()?; + self.post( + &json!({"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": params}), + &headers, + ) + .await? + .ok() + } +} + +/// Call `tool` on its tool server for the session `memory` is, through `egress`. A +/// server is asked once what it speaks (`server/discover`): one of 2026-07-28 is +/// called statelessly, one before it after the `initialize` handshake. /// /// # Errors /// @@ -180,76 +386,27 @@ pub async fn call( .and_then(|g| g.hosted.clone()) .ok_or_else(|| format!("[taps.{tool}] names no tool server"))?; let (_, path) = split_url(&url)?; - let post = |body: Value, session: Option| { - let path = path.clone(); - async move { - let mut headers = axum::http::HeaderMap::new(); - headers.insert( - "content-type", - "application/json".parse().map_err(|_| "header")?, - ); - headers.insert( - "accept", - "application/json, text/event-stream" - .parse() - .map_err(|_| "header")?, - ); - if let Some(s) = session { - headers.insert( - "mcp-session-id", - s.parse() - .map_err(|_| "the tool server's session id is not a header")?, - ); - } - let r = egress - .post(tool, &path, headers, body.to_string().into()) - .await - .map_err(|e| e.to_string())?; - if !(200..300).contains(&r.status) { - return Err(format!( - "the tool server answered {}: {}", - r.status, - r.body.trim() - )); - } - let content_type = r - .headers - .get("content-type") - .and_then(|v| v.to_str().ok()) - .unwrap_or_default() - .to_string(); - let id = r - .headers - .get("mcp-session-id") - .and_then(|v| v.to_str().ok()) - .map(str::to_string); - Ok::<_, String>((content_type, id, r.body)) + let server = Server { egress, tool, path }; + let era = match known(&url) { + Some(e) => e, + None => { + let e = server.discover().await?; + learn(&url, Some(e)); + e } }; - let (_, session, _) = post( - json!({ - "jsonrpc": "2.0", "id": 1, "method": "initialize", - "params": {"protocolVersion": "2025-06-18", "capabilities": {}, - "clientInfo": {"name": "amem", "version": env!("CARGO_PKG_VERSION")}} - }), - None, - ) - .await?; - post( - json!({"jsonrpc": "2.0", "method": "notifications/initialized"}), - session.clone(), - ) - .await?; let hosts = memory.fetch_hosts(tool); - let (content_type, _, body) = post( - json!({ - "jsonrpc": "2.0", "id": 2, "method": "tools/call", - "params": {"name": tool, "arguments": arguments(args, config, &hosts)} - }), - session, + let params = json!({"name": tool, "arguments": arguments(args, config, &hosts)}); + let sent = match era { + Era::Stateless => server.stateless(params).await, + Era::Handshake => server.handshake(params).await, + } + .inspect_err(|_| learn(&url, None))?; + checked( + &memory, + tool, + reply(&message(&sent.content_type, &sent.body)?)?, ) - .await?; - checked(&memory, tool, reply(&message(&content_type, &body)?)?) } #[cfg(test)] @@ -275,6 +432,14 @@ mod tests { assert_eq!(reply(&failed).unwrap_err(), "refused: x"); assert!(reply(&json!({"error": {"code": -32601}})).is_err()); assert!(reply(&json!({"result": {"content": [{"type": "text", "text": "{}"}]}})).is_err()); + // a result that says it is complete is one; any other kind is not taken + let mut done = ok.clone(); + done["result"]["resultType"] = json!("complete"); + assert_eq!(reply(&done).unwrap(), r); + let mut more = ok.clone(); + more["result"]["resultType"] = json!("input_required"); + let e = reply(&more).unwrap_err(); + assert!(e.contains("input_required"), "{e}"); assert!(message("application/json", "not json").is_err()); assert!(message("text/event-stream", "data: nothing\n").is_err()); let a = arguments("{\"url\": \"u\"}", "not json", &["h".into()]); @@ -283,6 +448,43 @@ mod tests { assert_eq!(a["hosts"][0], "h"); } + #[test] + fn what_a_server_speaks_is_read_from_its_answer_to_discovery() { + let result = |v: Value| { + json!({"jsonrpc": "2.0", "id": 1, + "result": {"resultType": "complete", "supportedVersions": v}}) + .to_string() + }; + let error = |code: i64, data: Value| { + json!({"jsonrpc": "2.0", "id": 1, "error": {"code": code, "message": "m", "data": data}}) + .to_string() + }; + let j = "application/json"; + assert_eq!(era(j, &result(json!(["2026-07-28"]))), Ok(Era::Stateless)); + let sse = format!("data: {}\n\n", result(json!(["2027-01-01", "2026-07-28"]))); + assert_eq!(era("text/event-stream", &sse), Ok(Era::Stateless)); + assert_eq!(era(j, &result(json!(["2025-11-25"]))), Ok(Era::Handshake)); + // refused for its version, naming only revisions with a handshake + let older = json!({"supported": ["2025-06-18"], "requested": "2026-07-28"}); + assert_eq!(era(j, &error(-32022, older)), Ok(Era::Handshake)); + // a server from before: a method unknown, nothing, a refusal in words, no list + assert_eq!(era(j, &error(-32601, Value::Null)), Ok(Era::Handshake)); + assert_eq!(era(j, ""), Ok(Era::Handshake)); + assert_eq!( + era("text/plain", "Bad Request: Session not found"), + Ok(Era::Handshake) + ); + let bare = json!({"jsonrpc": "2.0", "id": 1, "result": {}}).to_string(); + assert_eq!(era(j, &bare), Ok(Era::Handshake)); + // 2026-07-28 refusing the request itself, or versions none of which are spoken + let e = era(j, &error(-32020, Value::Null)).unwrap_err(); + assert!(e.contains("refused"), "{e}"); + let e = era(j, &result(json!(["2027-01-01"]))).unwrap_err(); + assert!(e.contains("2027-01-01") && e.contains("2026-07-28"), "{e}"); + let e = era(j, &error(-32022, json!({"supported": ["2027-01-01"]}))).unwrap_err(); + assert!(e.contains("2027-01-01"), "{e}"); + } + #[test] fn a_hosted_url_is_a_destination_and_a_path() { assert_eq!( diff --git a/crates/memory/src/main.rs b/crates/memory/src/main.rs index 0796de3e..3fff3a6a 100644 --- a/crates/memory/src/main.rs +++ b/crates/memory/src/main.rs @@ -27,6 +27,7 @@ mod hosted; mod image; mod librarian; mod mcp_http; +mod mcp_meta; mod mine; mod model; mod open; diff --git a/crates/memory/src/mcp_http.rs b/crates/memory/src/mcp_http.rs index bba31e5f..1cd0cf84 100644 --- a/crates/memory/src/mcp_http.rs +++ b/crates/memory/src/mcp_http.rs @@ -1,8 +1,10 @@ //! A small client for the memory server over MCP's streamable HTTP transport, for the -//! hook that runs inside the sandbox: one tool call, three requests, no session kept. -//! Plain HTTP/1.1 over a socket, since inside the sandbox the only server is the relay -//! on loopback and the only thing to say is one `tools/call`. +//! hook that runs inside the sandbox: one tool call, one stateless request of MCP +//! 2026-07-28, no handshake and no session. Plain HTTP/1.1 over a socket, since inside +//! the sandbox the only server is the relay on loopback, which passes bytes through to +//! this project's own memory server, and the only thing to say is one `tools/call`. +use crate::mcp_meta; use serde_json::{json, Value}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpStream; @@ -57,17 +59,18 @@ fn dechunk(raw: &str) -> String { /// POST `body` as JSON to `url` and read the whole answer; the connection is closed /// after it, so the end of the stream is the end of the answer. -async fn post(url: &str, session: Option<&str>, body: &str) -> Result { +async fn post(url: &str, headers: &[(&str, String)], body: &str) -> Result { let (authority, path) = split(url)?; let mut s = TcpStream::connect(&authority) .await .map_err(|e| format!("{authority}: {e}"))?; - let session = session - .map(|id| format!("mcp-session-id: {id}\r\n")) - .unwrap_or_default(); + let extra: String = headers + .iter() + .map(|(name, value)| format!("{name}: {value}\r\n")) + .collect(); let req = format!( "POST {path} HTTP/1.1\r\nHost: {authority}\r\nContent-Type: application/json\r\n\ - Accept: application/json, text/event-stream\r\n{session}Content-Length: {}\r\n\ + Accept: application/json, text/event-stream\r\n{extra}Content-Length: {}\r\n\ Connection: close\r\n\r\n{body}", body.len() ); @@ -123,46 +126,35 @@ fn messages(answer: &Answer) -> Vec { } } -/// Call one tool on the memory server at `url`, the MCP endpoint, as a session of its -/// own. The tool's result on success; on a refusal, the server's words. +/// Call one tool on the memory server at `url`, the MCP endpoint, in one stateless +/// request. The tool's result on success; on a refusal, the server's words. /// /// # Errors /// /// The server unreachable, a protocol answer that is not one, or the tool's error. pub async fn call_tool(url: &str, tool: &str, arguments: Value) -> Result { - let init = json!({ - "jsonrpc": "2.0", "id": 1, "method": "initialize", - "params": { - "protocolVersion": "2025-06-18", "capabilities": {}, - "clientInfo": { "name": "memory-hook", "version": env!("CARGO_PKG_VERSION") } - } - }); - let opened = post(url, None, &init.to_string()).await?; - if !(200..300).contains(&opened.status) { - return Err(format!("initialize answered {}", opened.status)); - } - let session = opened.header("mcp-session-id").map(str::to_string); - let ready = json!({ "jsonrpc": "2.0", "method": "notifications/initialized" }); - post(url, session.as_deref(), &ready.to_string()).await?; - let call = json!({ - "jsonrpc": "2.0", "id": 2, "method": "tools/call", - "params": { "name": tool, "arguments": arguments } - }); - let answered = post(url, session.as_deref(), &call.to_string()).await?; + let call = mcp_meta::request( + 1, + "tools/call", + json!({ "name": tool, "arguments": arguments }), + "memory-hook", + ); + let answered = post(url, &mcp_meta::headers(&call), &call.to_string()).await?; + let reply = messages(&answered).into_iter().find(|m| m["id"] == 1); if !(200..300).contains(&answered.status) { - return Err(format!( - "{tool} answered {}: {}", - answered.status, answered.body - )); + // a request refused as such comes back with its JSON-RPC error in the body + let why = reply + .as_ref() + .and_then(|m| m["error"]["message"].as_str()) + .map_or_else(|| answered.body.clone(), str::to_string); + return Err(format!("{tool} answered {}: {why}", answered.status)); } - let reply = messages(&answered) - .into_iter() - .find(|m| m["id"] == 2) - .ok_or_else(|| format!("{tool}: no reply in {:?}", answered.body))?; + let reply = reply.ok_or_else(|| format!("{tool}: no reply in {:?}", answered.body))?; if let Some(err) = reply.get("error") { return Err(err["message"].as_str().unwrap_or("error").to_string()); } let result = reply["result"].clone(); + mcp_meta::complete(&result).map_err(|e| format!("{tool}: {e}"))?; if result["isError"] == true { let text = result["content"][0]["text"].as_str().unwrap_or("refused"); return Err(text.to_string()); @@ -219,6 +211,25 @@ mod tests { let got = call_tool(&url, "get", json!({ "id": w.id })).await.unwrap(); let text = got.to_string(); assert!(text.contains("backoff"), "{text}"); + // the call was one stateless 2026-07-28 request: the server says its result is + // complete, which it says only to a request of that revision + assert_eq!(got["resultType"], "complete", "{text}"); + // and it answers discovery without a session, naming that revision + let discover = mcp_meta::request(1, "server/discover", Value::Null, "test"); + let answered = post(&url, &mcp_meta::headers(&discover), &discover.to_string()) + .await + .unwrap(); + assert_eq!(answered.status, 200, "{}", answered.body); + let found = messages(&answered).into_iter().find(|m| m["id"] == 1); + let versions = found.map(|m| m["result"]["supportedVersions"].clone()); + assert!( + versions + .as_ref() + .and_then(Value::as_array) + .is_some_and(|v| v.contains(&json!("2026-07-28"))), + "{versions:?}: {}", + answered.body + ); let err = call_tool(&url, "get", json!({ "id": "mem-me-nothere" })) .await .unwrap_err(); diff --git a/crates/memory/src/mcp_meta.rs b/crates/memory/src/mcp_meta.rs new file mode 100644 index 00000000..00a3f488 --- /dev/null +++ b/crates/memory/src/mcp_meta.rs @@ -0,0 +1,203 @@ +//! What MCP 2026-07-28 asks of a request and a result, for the hand-written client and +//! server in this binary (the hook's client, the hosted-tool client, `amem tool-server`): +//! no handshake, so every request names its protocol version and the client's +//! capabilities in `_meta`; over streamable HTTP it mirrors its method, and the name it +//! calls, into headers; every result says what kind it is in `resultType`. +//! +//! The revisions before it opened a session with `initialize`; each piece here still +//! speaks that for a counterpart that knows nothing newer. + +use base64::Engine as _; +use serde_json::{json, Value}; + +/// The revision without a handshake, which every request names in its `_meta`. +pub const MODERN: &str = "2026-07-28"; + +/// The revisions with an `initialize` handshake still answered or spoken, newest first. +pub const LEGACY: [&str; 3] = ["2025-11-25", "2025-06-18", "2025-03-26"]; + +/// The `_meta` key naming a request's protocol version. +pub const PROTOCOL_VERSION: &str = "io.modelcontextprotocol/protocolVersion"; +/// The `_meta` key carrying the client's capabilities for a request. +pub const CLIENT_CAPABILITIES: &str = "io.modelcontextprotocol/clientCapabilities"; +/// The `_meta` key naming the client. +pub const CLIENT_INFO: &str = "io.modelcontextprotocol/clientInfo"; +/// The `_meta` key naming the server on a result. +pub const SERVER_INFO: &str = "io.modelcontextprotocol/serverInfo"; + +/// The header mirroring the request's protocol version. +pub const VERSION_HEADER: &str = "mcp-protocol-version"; +/// The header mirroring the request's method. +pub const METHOD_HEADER: &str = "mcp-method"; +/// The header mirroring the tool, prompt, or resource a request names. +pub const NAME_HEADER: &str = "mcp-name"; + +/// The headers do not match the body, or one the request needs is missing. +pub const HEADER_MISMATCH: i64 = -32020; +/// The request names a protocol version the server does not speak. +pub const UNSUPPORTED_VERSION: i64 = -32022; +/// The errors only a server of 2026-07-28 or later answers with; a client that sees +/// one is talking to such a server and does not fall back to the handshake. +pub const MODERN_ERRORS: [i64; 3] = [HEADER_MISMATCH, -32021, UNSUPPORTED_VERSION]; + +/// Every version spoken, newest first. +#[must_use] +pub fn supported() -> Vec<&'static str> { + std::iter::once(MODERN).chain(LEGACY).collect() +} + +/// The methods whose request names a tool, prompt, or resource, which goes in the +/// `Mcp-Name` header, and the parameter it is read from. +#[must_use] +pub fn named_by(method: &str) -> Option<&'static str> { + match method { + "tools/call" | "prompts/get" => Some("name"), + "resources/read" => Some("uri"), + _ => None, + } +} + +/// A 2026-07-28 request from the client `client`: `params` with the `_meta` every +/// request carries. This client asks for no optional capability. +#[must_use] +pub fn request(id: u64, method: &str, mut params: Value, client: &str) -> Value { + if !params.is_object() { + params = json!({}); + } + params["_meta"] = json!({ + PROTOCOL_VERSION: MODERN, + CLIENT_CAPABILITIES: {}, + CLIENT_INFO: { "name": client, "version": env!("CARGO_PKG_VERSION") }, + }); + json!({ "jsonrpc": "2.0", "id": id, "method": method, "params": params }) +} + +/// The headers a streamable HTTP POST of `request` carries: its version, its method, +/// and, for a method that names something, that name. +#[must_use] +pub fn headers(request: &Value) -> Vec<(&'static str, String)> { + let method = request["method"].as_str().unwrap_or_default(); + let mut out = vec![ + (VERSION_HEADER, MODERN.to_string()), + (METHOD_HEADER, encode(method)), + ]; + if let Some(key) = named_by(method) { + if let Some(name) = request["params"][key].as_str() { + out.push((NAME_HEADER, encode(name))); + } + } + out +} + +const SENTINEL: (&str, &str) = ("=?base64?", "?="); + +/// A value as a header carries it: as is when it is visible ASCII with no space at +/// either end, else its UTF-8 in base64 between `=?base64?` and `?=`, as is a value +/// that already looks like that. +#[must_use] +pub fn encode(value: &str) -> String { + let plain = !value.is_empty() + && value.bytes().all(|b| (0x20..=0x7e).contains(&b)) + && !value.starts_with(' ') + && !value.ends_with(' ') + && !(value.starts_with(SENTINEL.0) && value.ends_with(SENTINEL.1)); + if plain { + value.to_string() + } else { + let b64 = base64::engine::general_purpose::STANDARD.encode(value); + format!("{}{b64}{}", SENTINEL.0, SENTINEL.1) + } +} + +/// A header value as [`encode`] wrote it, decoded; `None` when the base64 in it does +/// not decode to UTF-8. +#[must_use] +pub fn decode(header: &str) -> Option { + match header + .strip_prefix(SENTINEL.0) + .and_then(|h| h.strip_suffix(SENTINEL.1)) + { + Some(b64) => base64::engine::general_purpose::STANDARD + .decode(b64) + .ok() + .and_then(|b| String::from_utf8(b).ok()), + None => Some(header.to_string()), + } +} + +/// Whether a result is a final one. A server before 2026-07-28 sends no `resultType`, +/// which counts as complete; a kind this client does not take is refused. +/// +/// # Errors +/// +/// A `resultType` other than `complete`, named. +pub fn complete(result: &Value) -> Result<(), String> { + match result.get("resultType") { + None => Ok(()), + Some(Value::String(t)) if t == "complete" => Ok(()), + Some(other) => Err(format!( + "the server answered a {other} result, and only a complete one is taken" + )), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_request_carries_its_version_and_capabilities_in_meta_and_headers() { + let r = request( + 7, + "tools/call", + json!({"name": "get", "arguments": {}}), + "t", + ); + assert_eq!(r["params"]["_meta"][PROTOCOL_VERSION], MODERN); + assert!(r["params"]["_meta"][CLIENT_CAPABILITIES].is_object()); + assert_eq!(r["params"]["_meta"][CLIENT_INFO]["name"], "t"); + assert_eq!(r["params"]["name"], "get"); + assert_eq!( + headers(&r), + vec![ + (VERSION_HEADER, MODERN.to_string()), + (METHOD_HEADER, "tools/call".to_string()), + (NAME_HEADER, "get".to_string()), + ] + ); + let d = request(1, "server/discover", Value::Null, "t"); + assert_eq!(d["params"]["_meta"][PROTOCOL_VERSION], MODERN); + assert_eq!(headers(&d).len(), 2); + } + + #[test] + fn a_value_unsafe_in_a_header_goes_as_base64_and_comes_back() { + assert_eq!(encode("fetch"), "fetch"); + for v in [ + "Hello, 世界", + " padded ", + "line1\nline2", + "=?base64?literal?=", + "", + ] { + let e = encode(v); + assert!( + e.starts_with("=?base64?") && e.ends_with("?="), + "{v:?}: {e}" + ); + assert_eq!(decode(&e).as_deref(), Some(v)); + } + assert_eq!(encode("Hello, 世界"), "=?base64?SGVsbG8sIOS4lueVjA==?="); + assert_eq!(decode("plain").as_deref(), Some("plain")); + assert_eq!(decode("=?base64?!!?="), None); + } + + #[test] + fn a_result_without_a_kind_is_complete_and_an_unknown_kind_is_refused() { + assert!(complete(&json!({"content": []})).is_ok()); + assert!(complete(&json!({"resultType": "complete"})).is_ok()); + let e = complete(&json!({"resultType": "input_required"})).unwrap_err(); + assert!(e.contains("input_required"), "{e}"); + assert!(complete(&json!({"resultType": 3})).is_err()); + } +} diff --git a/crates/memory/src/tool_server.rs b/crates/memory/src/tool_server.rs index d309b856..f605b2d5 100644 --- a/crates/memory/src/tool_server.rs +++ b/crates/memory/src/tool_server.rs @@ -14,6 +14,10 @@ //! Off loopback the server wants a bearer token, so it is never an open proxy; in the //! Foundation's deployment it sits behind the gateway's MCP registry. +use crate::mcp_meta::{ + decode, named_by, supported, CLIENT_CAPABILITIES, HEADER_MISMATCH, LEGACY, METHOD_HEADER, + MODERN, NAME_HEADER, PROTOCOL_VERSION, SERVER_INFO, UNSUPPORTED_VERSION, VERSION_HEADER, +}; use crate::Fallible; use axum::extract::State; use axum::http::{HeaderMap, StatusCode}; @@ -59,20 +63,219 @@ fn served(s: &Served) -> Vec<(&String, &TapGrant)> { .collect() } -fn rpc(id: &Value, result: Value) -> Response { +/// Which revision a request is served under. One that names its version in `_meta`, +/// or asks `server/discover`, which only 2026-07-28 has, or mirrors a version newer +/// than the handshake's in its header, is served statelessly by 2026-07-28; anything +/// else as the revisions with `initialize` answer it, as this server always has. +#[derive(Clone, Copy, PartialEq, Eq)] +enum Era { + Handshake, + Stateless, +} + +fn era(method: &str, params: &Value, headers: &HeaderMap) -> Era { + if method == "initialize" { + return Era::Handshake; + } + let newer = match header(headers, VERSION_HEADER) { + Ok(v) => v.is_some_and(|v| !LEGACY.contains(&v.as_str())), + Err(_) => true, + }; + if params["_meta"].get(PROTOCOL_VERSION).is_some() || method == "server/discover" || newer { + Era::Stateless + } else { + Era::Handshake + } +} + +/// The one value of header `name`, if it was sent; more than one, or one that is not +/// text, is an error. +fn header(headers: &HeaderMap, name: &str) -> Result, Refusal> { + let mut all = headers.get_all(name).iter(); + let Some(first) = all.next() else { + return Ok(None); + }; + if all.next().is_some() { + return Err(Refusal::mismatch(format!("{name} is sent more than once"))); + } + first + .to_str() + .map(|v| Some(v.to_string())) + .map_err(|_| Refusal::mismatch(format!("{name} is not text"))) +} + +/// A request refused, before it is served. +struct Refusal { + code: i64, + message: String, + data: Option, +} + +impl Refusal { + fn mismatch(message: String) -> Refusal { + Refusal { + code: HEADER_MISMATCH, + message, + data: None, + } + } + + fn invalid(message: String) -> Refusal { + Refusal { + code: -32602, + message, + data: None, + } + } + + fn unsupported(requested: &str) -> Refusal { + Refusal { + code: UNSUPPORTED_VERSION, + message: format!( + "Unsupported protocol version {requested}; this server speaks {}", + supported().join(", ") + ), + data: Some(json!({"supported": supported(), "requested": requested})), + } + } + + /// The JSON-RPC error. A stateless request learns what kind from the status too: + /// malformed, at odds with its headers, or of a version not spoken is a 400, a + /// method unknown a 404. After a handshake every error went with a 200 as before, + /// but headers at odds with the body, which are a 400 whoever sends them. + fn answer(self, era: Era, id: &Value) -> Response { + let mut error = json!({"code": self.code, "message": self.message}); + if let Some(data) = self.data { + error["data"] = data; + } + let status = match (era, self.code) { + (_, HEADER_MISMATCH) + | (Era::Stateless, -32602 | -32600 | -32021 | UNSUPPORTED_VERSION) => { + StatusCode::BAD_REQUEST + } + (Era::Stateless, -32601) => StatusCode::NOT_FOUND, + _ => StatusCode::OK, + }; + ( + status, + Json(json!({"jsonrpc": "2.0", "id": id, "error": error})), + ) + .into_response() + } +} + +fn server_info() -> Value { + json!({"name": "amem tool-server", "version": env!("CARGO_PKG_VERSION")}) +} + +/// A result. To a stateless request it says it is complete and names this server, +/// since no handshake did. +fn answer(era: Era, id: &Value, mut result: Value) -> Response { + if era == Era::Stateless { + result["resultType"] = json!("complete"); + result["_meta"] = json!({ SERVER_INFO: server_info() }); + } Json(json!({"jsonrpc": "2.0", "id": id, "result": result})).into_response() } -fn rpc_error(id: &Value, code: i64, message: &str) -> Response { - Json(json!({"jsonrpc": "2.0", "id": id, "error": {"code": code, "message": message}})) - .into_response() +fn tool_error(message: &str) -> Value { + json!({"isError": true, "content": [{"type": "text", "text": message}]}) } -fn tool_error(id: &Value, message: &str) -> Response { - rpc( - id, - json!({"isError": true, "content": [{"type": "text", "text": message}]}), - ) +/// What 2026-07-28 asks of a request before it is served: a version this server speaks +/// in `_meta`, mirrored in its header; the client's capabilities; the method, and the +/// name a call names, mirrored in theirs. A header missing is refused as one at odds: +/// that revision requires them, and something between may route on them. +fn checked(method: &str, params: &Value, headers: &HeaderMap) -> Result<(), Refusal> { + let meta = ¶ms["_meta"]; + let sent = header(headers, VERSION_HEADER)?; + let Some(version) = meta[PROTOCOL_VERSION].as_str() else { + return Err(match sent.filter(|v| !supported().contains(&v.as_str())) { + Some(v) => Refusal::unsupported(&v), + None => Refusal::invalid(format!("the request's _meta names no {PROTOCOL_VERSION}")), + }); + }; + if version != MODERN { + return Err(Refusal::unsupported(version)); + } + match sent { + Some(v) if v == version => {} + Some(v) => { + return Err(Refusal::mismatch(format!( + "MCP-Protocol-Version {v} is not the {version} the body names" + ))) + } + None => return Err(Refusal::mismatch("MCP-Protocol-Version is missing".into())), + } + if !meta[CLIENT_CAPABILITIES].is_object() { + return Err(Refusal::invalid(format!( + "the request's _meta carries no {CLIENT_CAPABILITIES}" + ))); + } + if header(headers, METHOD_HEADER)?.is_none() { + return Err(Refusal::mismatch("Mcp-Method is missing".into())); + } + if let Some(key) = named_by(method) { + let named = params[key].as_str().unwrap_or_default(); + let sent = header(headers, NAME_HEADER)? + .ok_or_else(|| Refusal::mismatch("Mcp-Name is missing".into()))?; + if decode(&sent).as_deref() != Some(named) { + return Err(Refusal::mismatch(format!( + "Mcp-Name {sent} is not the {named} the body names" + ))); + } + } + Ok(()) +} + +/// The answer to `initialize`: the version asked when it is one with a handshake this +/// server speaks, else the newest of those, for the client to take or leave. +fn initialized(params: &Value) -> Value { + let asked = params["protocolVersion"].as_str().unwrap_or_default(); + let version = LEGACY + .iter() + .find(|v| **v == asked) + .copied() + .unwrap_or(LEGACY[0]); + json!({ + "protocolVersion": version, + "capabilities": {"tools": {}}, + "serverInfo": server_info(), + }) +} + +/// The answer to `server/discover`. Nothing in it is promised fresh for any time, and +/// with a token in front of the server none of it is for a shared cache. +fn discovered() -> Value { + json!({ + "supportedVersions": supported(), + "capabilities": {"tools": {}}, + "ttlMs": 0, + "cacheScope": "private", + }) +} + +/// The answer to `tools/list`. What is served follows what is installed under the root +/// at the time of asking, so it is not promised fresh either. +fn listed(s: &Served) -> Value { + json!({ + "tools": served(s).iter().map(|(name, _)| json!({ + "name": name, + "description": format!("The {name} tool, run under this server's grant; answers with the URLs it reached"), + "inputSchema": { + "type": "object", + "properties": { + "args": {"type": "object", "description": "the tool's own arguments"}, + "config": {"description": "the caller's configuration for the tool"}, + "hosts": {"type": "array", "items": {"type": "string"}, + "description": "the hosts the calling session may reach"}, + }, + "required": ["args", "hosts"], + }, + })).collect::>(), + "ttlMs": 0, + "cacheScope": "private", + }) } async fn mcp(State(s): State>, headers: HeaderMap, body: String) -> Response { @@ -86,7 +289,12 @@ async fn mcp(State(s): State>, headers: HeaderMap, body: String) -> } } let Ok(msg) = serde_json::from_str::(&body) else { - return rpc_error(&Value::Null, -32700, "not JSON"); + let refused = Refusal { + code: -32700, + message: "not JSON".into(), + data: None, + }; + return refused.answer(Era::Handshake, &Value::Null); }; let id = msg.get("id").cloned().unwrap_or(Value::Null); let method = msg["method"].as_str().unwrap_or_default(); @@ -94,45 +302,45 @@ async fn mcp(State(s): State>, headers: HeaderMap, body: String) -> // a notification: nothing to answer return StatusCode::ACCEPTED.into_response(); } + let params = &msg["params"]; + let era = era(method, params, &headers); + // a method header that names another method than the body is refused whoever sends + // it: whatever routed on the header acted on a different request than this one + let mirrored = header(&headers, METHOD_HEADER).and_then(|m| match m { + Some(m) if m != method => Err(Refusal::mismatch(format!( + "Mcp-Method {m} is not the {method} the body names" + ))), + _ => Ok(()), + }); + let ready = match era { + Era::Stateless => mirrored.and_then(|()| checked(method, params, &headers)), + Era::Handshake => mirrored, + }; + if let Err(refused) = ready { + return refused.answer(era, &id); + } match method { - "initialize" => rpc( - &id, - json!({ - "protocolVersion": msg["params"]["protocolVersion"].as_str().unwrap_or("2025-06-18"), - "capabilities": {"tools": {}}, - "serverInfo": {"name": "amem tool-server", "version": env!("CARGO_PKG_VERSION")}, - }), - ), - "tools/list" => rpc( - &id, - json!({"tools": served(&s).iter().map(|(name, _)| json!({ - "name": name, - "description": format!("The {name} tool, run under this server's grant; answers with the URLs it reached"), - "inputSchema": { - "type": "object", - "properties": { - "args": {"type": "object", "description": "the tool's own arguments"}, - "config": {"description": "the caller's configuration for the tool"}, - "hosts": {"type": "array", "items": {"type": "string"}, - "description": "the hosts the calling session may reach"}, - }, - "required": ["args", "hosts"], - }, - })).collect::>()}), - ), - "tools/call" => call(&s, &id, &msg["params"]).await, - _ => rpc_error(&id, -32601, &format!("no method {method}")), + "initialize" => answer(era, &id, initialized(params)), + "server/discover" => answer(era, &id, discovered()), + "tools/list" => answer(era, &id, listed(&s)), + "tools/call" => answer(era, &id, call(&s, params).await), + _ => Refusal { + code: -32601, + message: format!("no method {method}"), + data: None, + } + .answer(era, &id), } } -async fn call(s: &Served, id: &Value, params: &Value) -> Response { +async fn call(s: &Served, params: &Value) -> Value { let name = params["name"].as_str().unwrap_or_default(); let Some((_, grant)) = served(s).into_iter().find(|(n, _)| n.as_str() == name) else { - return tool_error(id, &format!("no tool {name} is served here")); + return tool_error(&format!("no tool {name} is served here")); }; let a = ¶ms["arguments"]; let Some(hosts) = a["hosts"].as_array() else { - return tool_error(id, "the call names no hosts its session may reach"); + return tool_error("the call names no hosts its session may reach"); }; let text = |v: &Value| match v { Value::String(t) => t.clone(), @@ -179,15 +387,17 @@ async fn call(s: &Served, id: &Value, params: &Value) -> Response { let answer = serde_json::from_str::(&done.answer).unwrap_or(Value::String(done.answer)); let body = json!({"answer": answer, "reached": done.reached}); - rpc( - id, - json!({"content": [{"type": "text", "text": body.to_string()}]}), - ) + json!({"content": [{"type": "text", "text": body.to_string()}]}) } - Err(e) => tool_error(id, &e.to_string()), + Err(e) => tool_error(&e.to_string()), } } +/// The MCP endpoint, at `/mcp`. +fn router(s: Arc) -> Router { + Router::new().route("/mcp", post(mcp)).with_state(s) +} + /// Whether an address to listen on is this machine's loopback. fn loopback(listen: &str) -> bool { listen @@ -239,14 +449,214 @@ pub async fn serve(root: &Path, listen: &str, token_env: Option<&str>) -> Fallib }, listener.local_addr()? ); - let app = Router::new().route("/mcp", post(mcp)).with_state(s); - axum::serve(listener, app).await?; + axum::serve(listener, router(s)).await?; Ok(()) } #[cfg(test)] mod tests { use super::*; + use crate::mcp_meta::{self, MODERN, PROTOCOL_VERSION}; + use tower::ServiceExt as _; + + /// A server for fetch, the binary's own, on a root with nothing installed. + fn app() -> Router { + let grants = Grants::parse( + "owner = \"human:test:me\"\n[domains.dom-ts]\n\ + [taps.fetch]\ndomain = \"dom-ts\"\ntool = true\nceiling = \"public\"\nhosts = [\"127.0.0.1\"]\n", + ) + .unwrap(); + router(Arc::new(Served { + root: PathBuf::from("/nonexistent"), + grants, + token: None, + })) + } + + /// POST `body` with `headers`; the status and the JSON answered, `null` for none. + async fn post_to(app: &Router, headers: &[(&str, String)], body: &Value) -> (u16, Value) { + let mut req = axum::http::Request::post("/mcp") + .header("content-type", "application/json") + .header("accept", "application/json, text/event-stream"); + for (k, v) in headers { + req = req.header(*k, v); + } + let resp = app + .clone() + .oneshot(req.body(axum::body::Body::from(body.to_string())).unwrap()) + .await + .unwrap(); + let status = resp.status().as_u16(); + let bytes = axum::body::to_bytes(resp.into_body(), 1 << 20) + .await + .unwrap(); + ( + status, + serde_json::from_slice(&bytes).unwrap_or(Value::Null), + ) + } + + /// A 2026-07-28 request and the headers it goes with. + fn modern(method: &str, params: Value) -> (Value, Vec<(&'static str, String)>) { + let r = mcp_meta::request(1, method, params, "test"); + let h = mcp_meta::headers(&r); + (r, h) + } + + fn call_fetch() -> Value { + json!({"name": "fetch", "arguments": {"args": {"url": "http://127.0.0.1:1/"}, "hosts": []}}) + } + + #[tokio::test] + async fn discovery_names_every_version_spoken_and_the_tools() { + let app = app(); + let (r, h) = modern("server/discover", Value::Null); + let (status, got) = post_to(&app, &h, &r).await; + assert_eq!(status, 200, "{got}"); + let result = &got["result"]; + assert_eq!(result["resultType"], "complete", "{got}"); + assert_eq!( + result["supportedVersions"], + json!(["2026-07-28", "2025-11-25", "2025-06-18", "2025-03-26"]), + "{got}" + ); + assert!(result["capabilities"]["tools"].is_object(), "{got}"); + assert_eq!( + result["_meta"][mcp_meta::SERVER_INFO]["name"], + "amem tool-server", + "{got}" + ); + } + + #[tokio::test] + async fn a_stateless_request_is_served_and_its_result_says_it_is_complete() { + let app = app(); + let (r, h) = modern("tools/list", Value::Null); + let (status, got) = post_to(&app, &h, &r).await; + assert_eq!(status, 200, "{got}"); + assert_eq!(got["result"]["resultType"], "complete", "{got}"); + assert_eq!(got["result"]["tools"][0]["name"], "fetch", "{got}"); + assert_eq!(got["result"]["ttlMs"], 0, "{got}"); + assert_eq!(got["result"]["cacheScope"], "private", "{got}"); + // a tool's refusal is a complete result too, marked as an error + let (r, h) = modern("tools/call", call_fetch()); + let (status, got) = post_to(&app, &h, &r).await; + assert_eq!(status, 200, "{got}"); + assert_eq!(got["result"]["resultType"], "complete", "{got}"); + assert_eq!(got["result"]["isError"], true, "{got}"); + // a name that is not plain in a header goes as base64 and is read back + let (r, mut h) = modern("tools/call", call_fetch()); + h.retain(|(k, _)| *k != mcp_meta::NAME_HEADER); + h.push((mcp_meta::NAME_HEADER, "=?base64?ZmV0Y2g=?=".into())); + let (status, got) = post_to(&app, &h, &r).await; + assert_eq!(status, 200, "{got}"); + } + + #[tokio::test] + async fn an_unsupported_version_is_refused_naming_the_supported_ones() { + let app = app(); + let (mut r, mut h) = modern("tools/list", Value::Null); + r["params"]["_meta"][PROTOCOL_VERSION] = json!("1900-01-01"); + h[0].1 = "1900-01-01".into(); + let (status, got) = post_to(&app, &h, &r).await; + assert_eq!(status, 400, "{got}"); + assert_eq!(got["error"]["code"], -32022, "{got}"); + assert_eq!(got["error"]["data"]["requested"], "1900-01-01", "{got}"); + assert_eq!( + got["error"]["data"]["supported"], + json!(mcp_meta::supported()), + "{got}" + ); + } + + /// Each way a 2026-07-28 request's headers fail its body, and what is answered. + #[tokio::test] + async fn headers_missing_or_at_odds_with_the_body_are_refused() { + let app = app(); + let refused = |status: u16, got: &Value, code: i64| { + assert_eq!(status, 400, "{got}"); + assert_eq!(got["error"]["code"], code, "{got}"); + assert_eq!(got["id"], 1, "{got}"); + }; + let without = |h: &[(&'static str, String)], name: &str| { + h.iter() + .filter(|(k, _)| *k != name) + .cloned() + .collect::>() + }; + let (r, h) = modern("tools/call", call_fetch()); + for name in [ + mcp_meta::VERSION_HEADER, + mcp_meta::METHOD_HEADER, + mcp_meta::NAME_HEADER, + ] { + let (status, got) = post_to(&app, &without(&h, name), &r).await; + refused(status, &got, mcp_meta::HEADER_MISMATCH); + } + let mut other = without(&h, mcp_meta::NAME_HEADER); + other.push((mcp_meta::NAME_HEADER, "search".into())); + let (status, got) = post_to(&app, &other, &r).await; + refused(status, &got, mcp_meta::HEADER_MISMATCH); + let mut other = without(&h, mcp_meta::METHOD_HEADER); + other.push((mcp_meta::METHOD_HEADER, "tools/list".into())); + let (status, got) = post_to(&app, &other, &r).await; + refused(status, &got, mcp_meta::HEADER_MISMATCH); + let mut other = without(&h, mcp_meta::VERSION_HEADER); + other.push((mcp_meta::VERSION_HEADER, "2025-06-18".into())); + let (status, got) = post_to(&app, &other, &r).await; + refused(status, &got, mcp_meta::HEADER_MISMATCH); + // the capabilities every request carries are not optional + let mut bare = r.clone(); + bare["params"]["_meta"] + .as_object_mut() + .unwrap() + .remove(mcp_meta::CLIENT_CAPABILITIES); + let (status, got) = post_to(&app, &h, &bare).await; + refused(status, &got, -32602); + // and a method this version does not have is not found + let (r, h) = modern("ping", Value::Null); + let (status, got) = post_to(&app, &h, &r).await; + assert_eq!(status, 404, "{got}"); + assert_eq!(got["error"]["code"], -32601, "{got}"); + } + + #[tokio::test] + async fn a_client_that_opens_with_initialize_is_answered_as_before() { + let app = app(); + let init = |v: &str| { + json!({"jsonrpc": "2.0", "id": 1, "method": "initialize", + "params": {"protocolVersion": v, "capabilities": {}, + "clientInfo": {"name": "t", "version": "0"}}}) + }; + let (status, got) = post_to(&app, &[], &init("2025-06-18")).await; + assert_eq!(status, 200, "{got}"); + assert_eq!(got["result"]["protocolVersion"], "2025-06-18", "{got}"); + assert!(got["result"].get("resultType").is_none(), "{got}"); + // a version it does not speak is answered with the newest it does + let (_, got) = post_to(&app, &[], &init("2024-11-05")).await; + assert_eq!(got["result"]["protocolVersion"], "2025-11-25", "{got}"); + let (_, got) = post_to(&app, &[], &init(MODERN)).await; + assert_eq!(got["result"]["protocolVersion"], "2025-11-25", "{got}"); + // a method header that contradicts the body is refused on this path too + let (status, got) = post_to( + &app, + &[(mcp_meta::METHOD_HEADER, "tools/call".into())], + &init("2025-06-18"), + ) + .await; + assert_eq!(status, 400, "{got}"); + assert_eq!(got["error"]["code"], mcp_meta::HEADER_MISMATCH, "{got}"); + let list = json!({"jsonrpc": "2.0", "id": 2, "method": "tools/list"}); + let (status, got) = post_to( + &app, + &[(mcp_meta::VERSION_HEADER, "2025-06-18".into())], + &list, + ) + .await; + assert_eq!(status, 200, "{got}"); + assert_eq!(got["result"]["tools"][0]["name"], "fetch", "{got}"); + assert!(got["result"].get("resultType").is_none(), "{got}"); + } #[test] fn only_loopback_goes_without_a_token() { diff --git a/crates/memory/tests/hosted.rs b/crates/memory/tests/hosted.rs index 3a8e916f..1d52c584 100644 --- a/crates/memory/tests/hosted.rs +++ b/crates/memory/tests/hosted.rs @@ -54,7 +54,8 @@ fn fetch_component() -> PathBuf { repo.join("taps/target/wasm32-wasip2/release/tap_fetch.wasm") } -/// Answer each connection with what `respond` makes of its request line, until dropped. +/// Answer each connection with what `respond` makes of its request: the request line, +/// then its headers lower-cased one to a line, then its body; until dropped. fn serve(respond: fn(&str, u16) -> String) -> u16 { let listener = TcpListener::bind("127.0.0.1:0").unwrap(); let port = listener.local_addr().unwrap().port(); @@ -66,18 +67,26 @@ fn serve(respond: fn(&str, u16) -> String) -> u16 { let mut first = String::new(); reader.read_line(&mut first).unwrap_or(0); let mut length = 0; + let mut headers = String::new(); loop { let mut line = String::new(); if reader.read_line(&mut line).unwrap_or(0) == 0 || line == "\r\n" { break; } - if let Some(v) = line.to_ascii_lowercase().strip_prefix("content-length:") { + let line = line.to_ascii_lowercase(); + if let Some(v) = line.strip_prefix("content-length:") { length = v.trim().parse().unwrap_or(0); } + headers.push_str(line.trim_end()); + headers.push('\n'); } let mut body = vec![0u8; length]; reader.read_exact(&mut body).ok(); - let line = format!("{}{}", first.trim(), String::from_utf8_lossy(&body)); + let line = format!( + "{}\n{headers}{}", + first.trim(), + String::from_utf8_lossy(&body) + ); let _ = s.write_all(respond(&line, port).as_bytes()); }); } @@ -328,6 +337,102 @@ fn a_server_that_names_nowhere_or_somewhere_unreachable_is_not_believed() { assert!(!sources.contains("trust me"), "{sources}"); } +/// An MCP endpoint of 2026-07-28 only: it answers `server/discover` and a stateless +/// `tools/call` whose headers mirror its body, and refuses the handshake as a server +/// without one does. +fn modern_only(request: &str, port: u16) -> String { + let refuse = |code: i64, message: &str| { + let body = serde_json::json!({"jsonrpc": "2.0", "id": 1, + "error": {"code": code, "message": message}}); + http("400 Bad Request", "application/json", "", &body.to_string()) + }; + let has = |h: &str| request.lines().any(|l| l == h); + if request.contains("\"initialize\"") { + return refuse(-32601, "no initialize here; this server speaks 2026-07-28"); + } + if !has("mcp-protocol-version: 2026-07-28") + || !request.contains("\"io.modelcontextprotocol/protocolVersion\":\"2026-07-28\"") + || !request.contains("\"io.modelcontextprotocol/clientCapabilities\":{}") + { + return refuse(-32602, "no 2026-07-28 _meta and header"); + } + let body = if request.contains("\"server/discover\"") && has("mcp-method: server/discover") { + serde_json::json!({"jsonrpc": "2.0", "id": 1, "result": {"resultType": "complete", + "supportedVersions": ["2026-07-28"], "capabilities": {"tools": {}}}}) + } else if request.contains("\"tools/call\"") + && has("mcp-method: tools/call") + && has("mcp-name: fetch") + { + let text = serde_json::json!({"answer": {"text": "modern and stateless"}, + "reached": [format!("http://127.0.0.1:{port}/page")]}) + .to_string(); + serde_json::json!({"jsonrpc": "2.0", "id": 1, "result": {"resultType": "complete", + "content": [{"type": "text", "text": text}]}}) + } else { + return refuse(-32020, "the headers do not mirror the body"); + }; + // answered as an event stream, as a server may + http( + "200 OK", + "text/event-stream", + "", + &format!("event: message\ndata: {body}\n\n"), + ) +} + +#[test] +fn a_server_of_the_stateless_revision_is_called_without_a_handshake() { + let dir = tempfile::tempdir().unwrap(); + let url = format!("http://127.0.0.1:{}/mcp", serve(modern_only)); + let root = launcher(dir.path(), "near", &url, "[\"127.0.0.1\"]"); + let answer = ok(&fetch(&root, "http://127.0.0.1:1/page", "")); + assert!(answer.contains("modern and stateless"), "{answer}"); +} + +/// An MCP endpoint of a revision before 2026-07-28: no `server/discover`, and a call +/// only in the session `initialize` opens, at the version it answered with. +fn handshake_only(request: &str, port: u16) -> String { + let has = |h: &str| request.lines().any(|l| l == h); + if request.contains("\"initialize\"") { + let body = r#"{"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"2025-06-18","capabilities":{"tools":{}},"serverInfo":{"name":"old"}}}"#; + return http( + "200 OK", + "application/json", + "Mcp-Session-Id: s-1\r\n", + body, + ); + } + if !has("mcp-session-id: s-1") { + return http( + "400 Bad Request", + "text/plain", + "", + "Bad Request: no session", + ); + } + if request.contains("notifications/initialized") { + return http("202 Accepted", "text/plain", "", ""); + } + if !request.contains("\"tools/call\"") || !has("mcp-protocol-version: 2025-06-18") { + return http("400 Bad Request", "text/plain", "", "Bad Request"); + } + let text = serde_json::json!({"answer": {"text": "from before the change"}, + "reached": [format!("http://127.0.0.1:{port}/page")]}) + .to_string(); + let body = serde_json::json!({"jsonrpc": "2.0", "id": 2, + "result": {"content": [{"type": "text", "text": text}]}}); + http("200 OK", "application/json", "", &body.to_string()) +} + +#[test] +fn a_server_from_before_the_stateless_revision_is_called_after_the_handshake() { + let dir = tempfile::tempdir().unwrap(); + let url = format!("http://127.0.0.1:{}/mcp", serve(handshake_only)); + let root = launcher(dir.path(), "near", &url, "[\"127.0.0.1\"]"); + let answer = ok(&fetch(&root, "http://127.0.0.1:1/page", "")); + assert!(answer.contains("from before the change"), "{answer}"); +} + #[test] fn a_hosted_grant_is_a_tool_served_over_https_or_loopback() { let dir = tempfile::tempdir().unwrap(); From 5976ff518841839e8bc1d168b3ad7703d4a207b2 Mon Sep 17 00:00:00 2001 From: Sander Striker Date: Sat, 31 Oct 2026 15:44:17 +0100 Subject: [PATCH 2/5] fix: close the gaps the adversarial pass found in the MCP 2026-07-28 change amem tool-server now compares a sent Mcp-Name header with the body on both paths, as it already did for Mcp-Method. Before, a call whose header named "search" and whose body named "fetch" ran fetch on the handshake path. A client of the handshake can still leave out both headers. amem tool-server now refuses a body that is not one JSON object, for example a batch, with HTTP 400 and -32600. Before, it answered 202 and dropped the body. A stateless request that names an unsupported version now gets -32022 with only 2026-07-28 in data.supported. The message says that older versions are reached through initialize. Before, the list included the handshake versions, also the version that the request named. The hosted-tool client now reads the HTTP status of the server/discover answer. It falls back to the handshake on 400, 404, 405, 415 or 422 without a 2026-07-28 error, or on a success with -32601. Any other answer is an error, and the client does not keep it as the revision of the server. Before, a 502 page, a 429 or a -32001 refusal made the client use the handshake for the life of the process. 415 and 422 are in the list because rmcp session servers send them for a request that they cannot read or that comes before initialize. New tests cover each fix. A new test also covers the per-server cache: two calls make one discovery request, and a failed call makes the next call ask again. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Bhj4Ck4y5iwZUsjpgQx8A2 --- crates/memory/src/hosted.rs | 262 ++++++++++++++++++++++++++----- crates/memory/src/tool_server.rs | 127 +++++++++++---- crates/memory/tests/hosted.rs | 5 + 3 files changed, 324 insertions(+), 70 deletions(-) diff --git a/crates/memory/src/hosted.rs b/crates/memory/src/hosted.rs index dde64625..92eade97 100644 --- a/crates/memory/src/hosted.rs +++ b/crates/memory/src/hosted.rs @@ -201,36 +201,58 @@ fn learn(url: &str, era: Option) { /// What a tool server's answer to `server/discover` says it speaks. A result listing /// 2026-07-28 is that revision; one listing only revisions with a handshake, or that -/// revision's refusal of the version naming only those, is one of them. Anything else -/// no server of 2026-07-28 answers, so it comes from one before it that has no such -/// method: an error status without one of that revision's errors, a method unknown, -/// nothing at all. +/// revision's refusal of the version naming only those, is one of them. A server from +/// before it has no such method: it answers with a status of a request it cannot take +/// ([`before`]) and none of that revision's errors, or with a success carrying +/// "method not found", as `amem tool-server` did. Anything else says nothing of the +/// revision, an error page, a limit, a refusal of the caller, and is not taken as one. /// /// # Errors /// -/// That revision's refusal of the request itself, or a list of versions none of -/// which this launcher speaks. -fn era(content_type: &str, body: &str) -> Result { - let Ok(m) = message(content_type, body) else { - return Ok(Era::Handshake); - }; - let listed = if let Some(e) = m.get("error") { - let code = e["code"].as_i64().unwrap_or_default(); - if !MODERN_ERRORS.contains(&code) { - return Ok(Era::Handshake); +/// That revision's refusal of the request itself, versions none of which this +/// launcher speaks, or an answer that names no revision, with its status. +fn era(status: u16, content_type: &str, body: &str) -> Result { + let m = message(content_type, body).ok(); + let error = m.as_ref().and_then(|m| m.get("error")); + let code = error.and_then(|e| e["code"].as_i64()); + if let (Some(e), Some(c)) = (error, code) { + if c == UNSUPPORTED_VERSION { + return named(&e["data"]["supported"], false); } - if code != UNSUPPORTED_VERSION { + if MODERN_ERRORS.contains(&c) { return Err(format!("the tool server refused server/discover: {e}")); } - &e["data"]["supported"] - } else { - &m["result"]["supportedVersions"] - }; - let Some(listed) = listed.as_array() else { - return Ok(Era::Handshake); - }; - let listed: Vec<&str> = listed.iter().filter_map(Value::as_str).collect(); - if listed.contains(&MODERN) && m.get("error").is_none() { + } + let listed = m.as_ref().map(|m| &m["result"]["supportedVersions"]); + match listed { + Some(v) if (200..300).contains(&status) && v.is_array() => return named(v, true), + _ if (200..300).contains(&status) && code == Some(-32601) => return Ok(Era::Handshake), + _ if before(status) => return Ok(Era::Handshake), + _ => {} + } + let said: String = body.trim().chars().take(300).collect(); + Err(format!( + "the tool server answered server/discover with {status} and named no MCP revision: {said}" + )) +} + +/// A status a server from before 2026-07-28 answers a method it has no such thing as +/// with: a bad request, not found, a method not allowed, and the 415 and 422 that rmcp's +/// session servers answer a request they cannot read or one before `initialize` with. +/// A refusal of the caller (401, 403), a limit (429), or a failure on the way (5xx) says +/// nothing of the revision. +fn before(status: u16) -> bool { + matches!(status, 400 | 404 | 405 | 415 | 422) +} + +/// The revision a list of versions leaves: 2026-07-28 when it is listed and +/// `stateless` allows it, else the handshake when a revision with one is listed. +fn named(listed: &Value, stateless: bool) -> Result { + let listed: Vec<&str> = listed + .as_array() + .map(|l| l.iter().filter_map(Value::as_str).collect()) + .unwrap_or_default(); + if stateless && listed.contains(&MODERN) { Ok(Era::Stateless) } else if listed.iter().any(|v| LEGACY.contains(v)) { Ok(Era::Handshake) @@ -313,7 +335,7 @@ impl Server<'_> { async fn discover(&self) -> Result { let request = mcp_meta::request(1, "server/discover", Value::Null, "amem"); let sent = self.post(&request, &mcp_meta::headers(&request)).await?; - era(&sent.content_type, &sent.body) + era(sent.status, &sent.content_type, &sent.body) } /// One `tools/call` of 2026-07-28: one request, no session. @@ -412,6 +434,144 @@ pub async fn call( #[cfg(test)] mod tests { use super::*; + use axum::extract::State; + use axum::http::StatusCode; + use axum::response::IntoResponse; + use std::io::Write as _; + use std::sync::atomic::{AtomicBool, AtomicU16, AtomicUsize, Ordering::SeqCst}; + + /// A tool server of 2026-07-28 on loopback, counting what it is asked. Discovery is + /// answered with `discover` when set, a status, a content type and a body, and with + /// the version list otherwise; a call fails with a 502 while `failing` is set; the + /// handshake is refused. + #[derive(Default)] + struct Fake { + discover: Mutex>, + failing: AtomicBool, + port: AtomicU16, + discovers: AtomicUsize, + handshakes: AtomicUsize, + calls: AtomicUsize, + } + + async fn fake(State(f): State>, body: String) -> axum::response::Response { + let m: Value = serde_json::from_str(&body).unwrap_or_default(); + match m["method"].as_str().unwrap_or_default() { + "server/discover" => { + f.discovers.fetch_add(1, SeqCst); + let set = *f.discover.lock().unwrap(); + if let Some((status, kind, text)) = set { + let status = StatusCode::from_u16(status).unwrap(); + return (status, [("content-type", kind)], text).into_response(); + } + axum::Json( + json!({"jsonrpc": "2.0", "id": 1, "result": {"resultType": "complete", + "supportedVersions": ["2026-07-28"], "capabilities": {"tools": {}}}}), + ) + .into_response() + } + "tools/call" => { + f.calls.fetch_add(1, SeqCst); + if f.failing.load(SeqCst) { + return (StatusCode::BAD_GATEWAY, "upstream down").into_response(); + } + let page = format!("http://127.0.0.1:{}/page", f.port.load(SeqCst)); + let text = json!({"answer": {"text": "ok"}, "reached": [page]}).to_string(); + axum::Json( + json!({"jsonrpc": "2.0", "id": 1, "result": {"resultType": "complete", + "content": [{"type": "text", "text": text}]}}), + ) + .into_response() + } + _ => { + f.handshakes.fetch_add(1, SeqCst); + (StatusCode::BAD_REQUEST, "no handshake here").into_response() + } + } + } + + /// A root whose fetch is hosted by a [`Fake`], the egress to it, and the fake. + async fn hosted_fetch(dir: &std::path::Path) -> (Egress, Arc, Arc) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let f = Arc::new(Fake::default()); + f.port.store(port, SeqCst); + let app = axum::Router::new() + .route("/mcp", axum::routing::post(fake)) + .with_state(f.clone()); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + let root = dir.join("m"); + Memory::init(&root, "human:test:me", "dom-me").unwrap(); + let mut g = std::fs::OpenOptions::new() + .append(true) + .open(root.join("grants.toml")) + .unwrap(); + write!( + g, + "\n[taps.fetch]\ndomain = \"dom-me\"\ntool = true\nceiling = \"public\"\nhosts = [\"127.0.0.1\"]\nloopback = true\nhosted = \"http://127.0.0.1:{port}/mcp\"\n" + ) + .unwrap(); + let memory = Arc::new(Memory::open(&root).unwrap().without_read_log()); + let egress = Egress::with_tool_servers( + memory.clone(), + HashMap::new(), + tool_servers(memory.grants()).unwrap(), + &dir.join("t.db"), + ) + .unwrap(); + (egress, memory, f) + } + + async fn fetch(egress: &Egress, memory: &Arc) -> Result { + let args = "{\"url\": \"http://127.0.0.1:1/page\"}"; + call(egress, memory.clone(), "fetch", "{}", args).await + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn what_a_server_speaks_is_asked_once_and_again_after_a_call_fails() { + let dir = tempfile::tempdir().unwrap(); + let (egress, memory, f) = hosted_fetch(dir.path()).await; + fetch(&egress, &memory).await.unwrap(); + fetch(&egress, &memory).await.unwrap(); + assert_eq!(f.discovers.load(SeqCst), 1); + assert_eq!(f.calls.load(SeqCst), 2); + f.failing.store(true, SeqCst); + let e = fetch(&egress, &memory).await.unwrap_err(); + assert!(e.contains("502"), "{e}"); + assert_eq!(f.discovers.load(SeqCst), 1); + f.failing.store(false, SeqCst); + fetch(&egress, &memory).await.unwrap(); + assert_eq!(f.discovers.load(SeqCst), 2); + assert_eq!(f.handshakes.load(SeqCst), 0); + } + + /// An answer to discovery that says nothing of the revision, an error page, a limit, + /// a refusal of the caller, is an error, and is not taken for a server from before. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn an_answer_to_discovery_that_names_no_revision_is_an_error_and_not_kept() { + let dir = tempfile::tempdir().unwrap(); + let (egress, memory, f) = hosted_fetch(dir.path()).await; + for (status, kind, body, says) in [ + (502, "text/html", "Bad Gateway", "502"), + (429, "text/plain", "slow down", "429"), + ( + 200, + "application/json", + r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32001,"message":"unauthorized"}}"#, + "unauthorized", + ), + ] { + *f.discover.lock().unwrap() = Some((status, kind, body)); + let e = fetch(&egress, &memory).await.unwrap_err(); + assert!(e.contains(says), "{status}: {e}"); + assert_eq!(f.handshakes.load(SeqCst), 0, "{status}: {e}"); + } + // none of them was kept as what the server speaks: the next answer is asked for + *f.discover.lock().unwrap() = None; + fetch(&egress, &memory).await.unwrap(); + assert_eq!(f.discovers.load(SeqCst), 4); + assert_eq!(f.handshakes.load(SeqCst), 0); + } #[test] fn a_reply_is_read_from_json_or_an_event_stream() { @@ -460,29 +620,49 @@ mod tests { .to_string() }; let j = "application/json"; - assert_eq!(era(j, &result(json!(["2026-07-28"]))), Ok(Era::Stateless)); + let ok = |kind: &str, body: &str| era(200, kind, body); + assert_eq!(ok(j, &result(json!(["2026-07-28"]))), Ok(Era::Stateless)); let sse = format!("data: {}\n\n", result(json!(["2027-01-01", "2026-07-28"]))); - assert_eq!(era("text/event-stream", &sse), Ok(Era::Stateless)); - assert_eq!(era(j, &result(json!(["2025-11-25"]))), Ok(Era::Handshake)); + assert_eq!(ok("text/event-stream", &sse), Ok(Era::Stateless)); + assert_eq!(ok(j, &result(json!(["2025-11-25"]))), Ok(Era::Handshake)); // refused for its version, naming only revisions with a handshake let older = json!({"supported": ["2025-06-18"], "requested": "2026-07-28"}); - assert_eq!(era(j, &error(-32022, older)), Ok(Era::Handshake)); - // a server from before: a method unknown, nothing, a refusal in words, no list - assert_eq!(era(j, &error(-32601, Value::Null)), Ok(Era::Handshake)); - assert_eq!(era(j, ""), Ok(Era::Handshake)); - assert_eq!( - era("text/plain", "Bad Request: Session not found"), - Ok(Era::Handshake) - ); - let bare = json!({"jsonrpc": "2.0", "id": 1, "result": {}}).to_string(); - assert_eq!(era(j, &bare), Ok(Era::Handshake)); + assert_eq!(era(400, j, &error(-32022, older)), Ok(Era::Handshake)); + // a server from before: "method not found" in a success, as amem tool-server + // answered, or a status of a request it cannot take with no error of 2026-07-28 + assert_eq!(ok(j, &error(-32601, Value::Null)), Ok(Era::Handshake)); + for status in [400, 404, 405, 415, 422] { + assert_eq!(era(status, j, ""), Ok(Era::Handshake), "{status}"); + let words = "Unexpected message, expect initialize request"; + assert_eq!(era(status, "text/plain", words), Ok(Era::Handshake)); + } // 2026-07-28 refusing the request itself, or versions none of which are spoken - let e = era(j, &error(-32020, Value::Null)).unwrap_err(); + let e = era(400, j, &error(-32020, Value::Null)).unwrap_err(); assert!(e.contains("refused"), "{e}"); - let e = era(j, &result(json!(["2027-01-01"]))).unwrap_err(); + let e = ok(j, &result(json!(["2027-01-01"]))).unwrap_err(); assert!(e.contains("2027-01-01") && e.contains("2026-07-28"), "{e}"); - let e = era(j, &error(-32022, json!({"supported": ["2027-01-01"]}))).unwrap_err(); + let e = era(400, j, &error(-32022, json!({"supported": ["2027-01-01"]}))).unwrap_err(); assert!(e.contains("2027-01-01"), "{e}"); + // an answer that names no revision: an error page, a limit, a refusal of the + // caller, a success that is no answer to discovery + let page = "".to_string() + &"Bad Gateway ".repeat(100) + ""; + let e = era(502, "text/html", &page).unwrap_err(); + assert!(e.contains("502") && e.len() < 500, "{e}"); + for (status, body) in [ + (429, "slow down".to_string()), + (401, "a bearer token is required".to_string()), + (403, String::new()), + (503, String::new()), + (200, error(-32001, Value::Null)), + ( + 200, + json!({"jsonrpc": "2.0", "id": 1, "result": {}}).to_string(), + ), + (202, String::new()), + ] { + let e = era(status, j, &body).unwrap_err(); + assert!(e.contains(&status.to_string()), "{e}"); + } } #[test] diff --git a/crates/memory/src/tool_server.rs b/crates/memory/src/tool_server.rs index f605b2d5..d20ea101 100644 --- a/crates/memory/src/tool_server.rs +++ b/crates/memory/src/tool_server.rs @@ -128,31 +128,33 @@ impl Refusal { } } + /// A request without a handshake that names a version other than the one spoken + /// that way. The versions it lists are the ones such a request may name; the older + /// ones are reached through `initialize`, which the message says. fn unsupported(requested: &str) -> Refusal { Refusal { code: UNSUPPORTED_VERSION, message: format!( - "Unsupported protocol version {requested}; this server speaks {}", - supported().join(", ") + "Unsupported protocol version {requested}: a request without a handshake speaks {MODERN}; {} are reached through initialize", + LEGACY.join(", ") ), - data: Some(json!({"supported": supported(), "requested": requested})), + data: Some(json!({"supported": [MODERN], "requested": requested})), } } /// The JSON-RPC error. A stateless request learns what kind from the status too: /// malformed, at odds with its headers, or of a version not spoken is a 400, a /// method unknown a 404. After a handshake every error went with a 200 as before, - /// but headers at odds with the body, which are a 400 whoever sends them. + /// but headers at odds with the body and a body that is not one request, which are + /// a 400 whoever sends them. fn answer(self, era: Era, id: &Value) -> Response { let mut error = json!({"code": self.code, "message": self.message}); if let Some(data) = self.data { error["data"] = data; } let status = match (era, self.code) { - (_, HEADER_MISMATCH) - | (Era::Stateless, -32602 | -32600 | -32021 | UNSUPPORTED_VERSION) => { - StatusCode::BAD_REQUEST - } + (_, HEADER_MISMATCH | -32600) + | (Era::Stateless, -32602 | -32021 | UNSUPPORTED_VERSION) => StatusCode::BAD_REQUEST, (Era::Stateless, -32601) => StatusCode::NOT_FOUND, _ => StatusCode::OK, }; @@ -215,13 +217,30 @@ fn checked(method: &str, params: &Value, headers: &HeaderMap) -> Result<(), Refu if header(headers, METHOD_HEADER)?.is_none() { return Err(Refusal::mismatch("Mcp-Method is missing".into())); } - if let Some(key) = named_by(method) { - let named = params[key].as_str().unwrap_or_default(); - let sent = header(headers, NAME_HEADER)? - .ok_or_else(|| Refusal::mismatch("Mcp-Name is missing".into()))?; - if decode(&sent).as_deref() != Some(named) { + if named_by(method).is_some() && header(headers, NAME_HEADER)?.is_none() { + return Err(Refusal::mismatch("Mcp-Name is missing".into())); + } + Ok(()) +} + +/// Whether the method and name headers, where sent, say what the body says. Whoever +/// sends them, a request whose headers name another method or tool than its body is +/// refused: whatever routed on the headers acted on a different request than this one. +/// A client of the handshake may leave them out. +fn mirrored(method: &str, params: &Value, headers: &HeaderMap) -> Result<(), Refusal> { + if let Some(m) = header(headers, METHOD_HEADER)? { + if m != method { return Err(Refusal::mismatch(format!( - "Mcp-Name {sent} is not the {named} the body names" + "Mcp-Method {m} is not the {method} the body names" + ))); + } + } + if let Some(sent) = header(headers, NAME_HEADER)? { + let named = named_by(method).and_then(|key| params[key].as_str()); + if named.is_none() || decode(&sent).as_deref() != named { + return Err(Refusal::mismatch(format!( + "Mcp-Name {sent} is not the {} the body names", + named.unwrap_or("nothing") ))); } } @@ -296,6 +315,15 @@ async fn mcp(State(s): State>, headers: HeaderMap, body: String) -> }; return refused.answer(Era::Handshake, &Value::Null); }; + if !msg.is_object() { + // one request a POST: a batch, or JSON that is no request, is not one + let refused = Refusal { + code: -32600, + message: "the body is not one JSON-RPC request".into(), + data: None, + }; + return refused.answer(Era::Handshake, &Value::Null); + } let id = msg.get("id").cloned().unwrap_or(Value::Null); let method = msg["method"].as_str().unwrap_or_default(); if msg.get("id").is_none() { @@ -304,17 +332,11 @@ async fn mcp(State(s): State>, headers: HeaderMap, body: String) -> } let params = &msg["params"]; let era = era(method, params, &headers); - // a method header that names another method than the body is refused whoever sends - // it: whatever routed on the header acted on a different request than this one - let mirrored = header(&headers, METHOD_HEADER).and_then(|m| match m { - Some(m) if m != method => Err(Refusal::mismatch(format!( - "Mcp-Method {m} is not the {method} the body names" - ))), - _ => Ok(()), - }); let ready = match era { - Era::Stateless => mirrored.and_then(|()| checked(method, params, &headers)), - Era::Handshake => mirrored, + Era::Stateless => { + mirrored(method, params, &headers).and_then(|()| checked(method, params, &headers)) + } + Era::Handshake => mirrored(method, params, &headers), }; if let Err(refused) = ready { return refused.answer(era, &id); @@ -562,11 +584,58 @@ mod tests { assert_eq!(status, 400, "{got}"); assert_eq!(got["error"]["code"], -32022, "{got}"); assert_eq!(got["error"]["data"]["requested"], "1900-01-01", "{got}"); - assert_eq!( - got["error"]["data"]["supported"], - json!(mcp_meta::supported()), - "{got}" - ); + // a stateless request is answered with the stateless versions only; the older + // ones are reached through initialize, which the message says + assert_eq!(got["error"]["data"]["supported"], json!([MODERN]), "{got}"); + let said = got["error"]["message"].as_str().unwrap_or_default(); + assert!(said.contains("initialize"), "{got}"); + // a version with a handshake named in _meta is no stateless version either + let (mut r, mut h) = modern("tools/list", Value::Null); + r["params"]["_meta"][PROTOCOL_VERSION] = json!("2025-06-18"); + h[0].1 = "2025-06-18".into(); + let (status, got) = post_to(&app, &h, &r).await; + assert_eq!(status, 400, "{got}"); + assert_eq!(got["error"]["code"], -32022, "{got}"); + assert_eq!(got["error"]["data"]["supported"], json!([MODERN]), "{got}"); + } + + #[tokio::test] + async fn a_body_that_is_not_one_request_is_refused_as_invalid() { + let app = app(); + let one = json!({"jsonrpc": "2.0", "id": 1, "method": "tools/list"}); + for body in [json!([one]), json!([]), json!("tools/list"), json!(7)] { + let (status, got) = post_to(&app, &[], &body).await; + assert_eq!(status, 400, "{body}: {got}"); + assert_eq!(got["error"]["code"], -32600, "{body}: {got}"); + assert_eq!(got["id"], Value::Null, "{body}: {got}"); + } + } + + /// A name header at odds with the body is refused whoever sends it; without one, a + /// client of the handshake is served as before. + #[tokio::test] + async fn a_name_header_at_odds_with_the_body_is_refused_after_a_handshake_too() { + let app = app(); + let call = + json!({"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": call_fetch()}); + let method = (mcp_meta::METHOD_HEADER, "tools/call".to_string()); + let other = (mcp_meta::NAME_HEADER, "search".to_string()); + let version = (mcp_meta::VERSION_HEADER, "2025-06-18".to_string()); + for headers in [ + vec![method.clone(), other.clone()], + vec![method.clone(), other.clone(), version.clone()], + vec![other.clone()], + ] { + let (status, got) = post_to(&app, &headers, &call).await; + assert_eq!(status, 400, "{headers:?}: {got}"); + assert_eq!(got["error"]["code"], mcp_meta::HEADER_MISMATCH, "{got}"); + } + let same = (mcp_meta::NAME_HEADER, "=?base64?ZmV0Y2g=?=".to_string()); + for headers in [vec![], vec![version, same]] { + let (status, got) = post_to(&app, &headers, &call).await; + assert_eq!(status, 200, "{headers:?}: {got}"); + assert_eq!(got["result"]["isError"], true, "{got}"); + } } /// Each way a 2026-07-28 request's headers fail its body, and what is answered. diff --git a/crates/memory/tests/hosted.rs b/crates/memory/tests/hosted.rs index 1d52c584..67711109 100644 --- a/crates/memory/tests/hosted.rs +++ b/crates/memory/tests/hosted.rs @@ -302,6 +302,11 @@ fn a_tool_server_without_the_component_fetches_natively_behind_the_same_gate() { /// An MCP endpoint that answers every call with a reply naming `reached`. fn liar(reached: &'static str) -> String { fn respond(line: &str, reached: &str) -> String { + // a server from before 2026-07-28 knows no discovery, as amem tool-server did not + if line.contains("\"server/discover\"") { + let body = r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32601,"message":"no method server/discover"}}"#; + return http("200 OK", "application/json", "", body); + } let body = if line.contains("\"initialize\"") { r#"{"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"2025-06-18","capabilities":{"tools":{}},"serverInfo":{"name":"liar"}}}"#.to_string() } else if line.contains("tools/call") { From f8c0f29768f84f3cd8a0e3ffd4ec61cd663cea33 Mon Sep 17 00:00:00 2001 From: Sander Striker Date: Mon, 2 Nov 2026 23:27:28 +0100 Subject: [PATCH 3/5] fix: check the discovery result type, forget a stale revision, send Mcp-Method as is Discovery now refuses a result that is not complete. Such a result does not go into the cache. A call now clears the cached revision when its reply does not parse, has an unsupported resultType, or has a protocol error. A tool error, an unknown tool or a server error keeps the cache. The client now sends Mcp-Method without base64. MCP 2026-07-28 permits the sentinel only in Mcp-Name and Mcp-Param headers. The server keeps the comparison of Mcp-Method as sent, and a new test holds this. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Bhj4Ck4y5iwZUsjpgQx8A2 --- crates/memory/src/hosted.rs | 140 ++++++++++++++++++++++++++----- crates/memory/src/mcp_meta.rs | 15 +++- crates/memory/src/tool_server.rs | 21 ++++- 3 files changed, 154 insertions(+), 22 deletions(-) diff --git a/crates/memory/src/hosted.rs b/crates/memory/src/hosted.rs index 92eade97..cee93380 100644 --- a/crates/memory/src/hosted.rs +++ b/crates/memory/src/hosted.rs @@ -175,7 +175,8 @@ enum Era { /// What each tool server was found to speak, by URL, for the life of this process: a /// client keeps that per server rather than asking before every call. A call that -/// fails on the way forgets it, so the next one asks again. +/// fails on the way, or is answered with a reply not of that revision ([`unspoken`]), +/// forgets it, so the next one asks again. fn eras() -> &'static Mutex> { static ERAS: OnceLock>> = OnceLock::new(); ERAS.get_or_init(Mutex::default) @@ -199,18 +200,20 @@ fn learn(url: &str, era: Option) { }; } -/// What a tool server's answer to `server/discover` says it speaks. A result listing -/// 2026-07-28 is that revision; one listing only revisions with a handshake, or that -/// revision's refusal of the version naming only those, is one of them. A server from -/// before it has no such method: it answers with a status of a request it cannot take -/// ([`before`]) and none of that revision's errors, or with a success carrying -/// "method not found", as `amem tool-server` did. Anything else says nothing of the -/// revision, an error page, a limit, a refusal of the caller, and is not taken as one. +/// What a tool server's answer to `server/discover` says it speaks. A complete result +/// listing 2026-07-28 is that revision, and one of another kind is no list at all; one +/// listing only revisions with a handshake, or that revision's refusal of the version +/// naming only those, is one of them. A server from before it has no such method: it +/// answers with a status of a request it cannot take ([`before`]) and none of that +/// revision's errors, or with a success carrying "method not found", as `amem +/// tool-server` did. Anything else says nothing of the revision, an error page, a limit, +/// a refusal of the caller, and is not taken as one. /// /// # Errors /// -/// That revision's refusal of the request itself, versions none of which this -/// launcher speaks, or an answer that names no revision, with its status. +/// That revision's refusal of the request itself, a result of a kind not taken, +/// versions none of which this launcher speaks, or an answer that names no revision, +/// with its status. fn era(status: u16, content_type: &str, body: &str) -> Result { let m = message(content_type, body).ok(); let error = m.as_ref().and_then(|m| m.get("error")); @@ -223,9 +226,16 @@ fn era(status: u16, content_type: &str, body: &str) -> Result { return Err(format!("the tool server refused server/discover: {e}")); } } - let listed = m.as_ref().map(|m| &m["result"]["supportedVersions"]); + let listed = m + .as_ref() + .map(|m| &m["result"]) + .filter(|r| r["supportedVersions"].is_array()); match listed { - Some(v) if (200..300).contains(&status) && v.is_array() => return named(v, true), + Some(r) if (200..300).contains(&status) => { + mcp_meta::complete(r) + .map_err(|e| format!("the tool server's answer to server/discover: {e}"))?; + return named(&r["supportedVersions"], true); + } _ if (200..300).contains(&status) && code == Some(-32601) => return Ok(Era::Handshake), _ if before(status) => return Ok(Era::Handshake), _ => {} @@ -265,6 +275,21 @@ fn named(listed: &Value, stateless: bool) -> Result { } } +/// Whether a reply to a call says the revision kept for its server is not the one it +/// speaks: no JSON-RPC result or error, a result of a kind not taken, or an error with +/// the request's form or protocol (a parse error, an invalid request, a method not +/// found, or one of 2026-07-28's own) rather than with the call. A tool's refusal, an +/// unknown tool, or a failure in the server says nothing of the revision. +fn unspoken(message: &Value) -> bool { + match (message.get("error"), message.get("result")) { + (Some(e), _) => e["code"] + .as_i64() + .is_none_or(|c| [-32700, -32600, -32601].contains(&c) || MODERN_ERRORS.contains(&c)), + (None, Some(r)) => !r.is_object() || mcp_meta::complete(r).is_err(), + (None, None) => true, + } +} + /// One POST to a tool server, as it answered. struct Sent { status: u16, @@ -424,11 +449,11 @@ pub async fn call( Era::Handshake => server.handshake(params).await, } .inspect_err(|_| learn(&url, None))?; - checked( - &memory, - tool, - reply(&message(&sent.content_type, &sent.body)?)?, - ) + let answered = message(&sent.content_type, &sent.body).inspect_err(|_| learn(&url, None))?; + if unspoken(&answered) { + learn(&url, None); + } + checked(&memory, tool, reply(&answered)?) } #[cfg(test)] @@ -442,11 +467,12 @@ mod tests { /// A tool server of 2026-07-28 on loopback, counting what it is asked. Discovery is /// answered with `discover` when set, a status, a content type and a body, and with - /// the version list otherwise; a call fails with a 502 while `failing` is set; the - /// handshake is refused. + /// the version list otherwise; a call fails with a 502 while `failing` is set, and + /// is answered with `answer` when set; the handshake is refused. #[derive(Default)] struct Fake { discover: Mutex>, + answer: Mutex>, failing: AtomicBool, port: AtomicU16, discovers: AtomicUsize, @@ -475,6 +501,11 @@ mod tests { if f.failing.load(SeqCst) { return (StatusCode::BAD_GATEWAY, "upstream down").into_response(); } + let set = *f.answer.lock().unwrap(); + if let Some((status, kind, text)) = set { + let status = StatusCode::from_u16(status).unwrap(); + return (status, [("content-type", kind)], text).into_response(); + } let page = format!("http://127.0.0.1:{}/page", f.port.load(SeqCst)); let text = json!({"answer": {"text": "ok"}, "reached": [page]}).to_string(); axum::Json( @@ -573,6 +604,77 @@ mod tests { assert_eq!(f.handshakes.load(SeqCst), 0); } + /// A result of a kind this client does not take is no list of versions, whatever it + /// carries, and is not kept as what the server speaks. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_discovery_result_of_a_kind_not_taken_is_refused_and_not_kept() { + let dir = tempfile::tempdir().unwrap(); + let (egress, memory, f) = hosted_fetch(dir.path()).await; + let more = r#"{"jsonrpc":"2.0","id":1,"result":{"resultType":"input_required","supportedVersions":["2026-07-28"],"inputRequests":{}}}"#; + *f.discover.lock().unwrap() = Some((200, "application/json", more)); + let e = fetch(&egress, &memory).await.unwrap_err(); + assert!(e.contains("input_required"), "{e}"); + assert_eq!(f.calls.load(SeqCst), 0); + *f.discover.lock().unwrap() = None; + fetch(&egress, &memory).await.unwrap(); + assert_eq!(f.discovers.load(SeqCst), 2); + } + + /// A reply that is not one of the revision kept for the server forgets it, so the + /// next call asks again; a refusal of the call itself, with a success, does not. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn a_reply_not_of_the_revision_kept_asks_again_and_a_refused_call_does_not() { + let dir = tempfile::tempdir().unwrap(); + let (egress, memory, f) = hosted_fetch(dir.path()).await; + fetch(&egress, &memory).await.unwrap(); + let j = "application/json"; + let unspoken = [ + (j, "not JSON"), + ("text/event-stream", "data: nothing\n\n"), + (j, r#"{"jsonrpc":"2.0","id":1}"#), + (j, r#"{"jsonrpc":"2.0","id":1,"result":"done"}"#), + ( + j, + r#"{"jsonrpc":"2.0","id":1,"result":{"resultType":"input_required","inputRequests":{}}}"#, + ), + ( + j, + r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32022,"message":"m","data":{"supported":["2027-01-01"]}}}"#, + ), + ( + j, + r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32601,"message":"no such method"}}"#, + ), + ( + j, + r#"{"jsonrpc":"2.0","id":1,"error":{"message":"no code"}}"#, + ), + ]; + for (n, (kind, body)) in unspoken.into_iter().enumerate() { + *f.answer.lock().unwrap() = Some((200, kind, body)); + assert!(fetch(&egress, &memory).await.is_err(), "{body}"); + *f.answer.lock().unwrap() = None; + fetch(&egress, &memory).await.unwrap(); + assert_eq!(f.discovers.load(SeqCst), n + 2, "{body}"); + } + let asked = f.discovers.load(SeqCst); + let refused = [ + r#"{"jsonrpc":"2.0","id":1,"result":{"resultType":"complete","isError":true,"content":[{"type":"text","text":"refused: x"}]}}"#, + r#"{"jsonrpc":"2.0","id":1,"result":{"resultType":"complete","content":[{"type":"text","text":"not the tool's JSON"}]}}"#, + r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32602,"message":"Unknown tool: fetch"}}"#, + r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32603,"message":"internal error"}}"#, + r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32001,"message":"unauthorized"}}"#, + ]; + for body in refused { + *f.answer.lock().unwrap() = Some((200, j, body)); + assert!(fetch(&egress, &memory).await.is_err(), "{body}"); + *f.answer.lock().unwrap() = None; + fetch(&egress, &memory).await.unwrap(); + assert_eq!(f.discovers.load(SeqCst), asked, "{body}"); + } + assert_eq!(f.handshakes.load(SeqCst), 0); + } + #[test] fn a_reply_is_read_from_json_or_an_event_stream() { let ok = json!({"jsonrpc": "2.0", "id": 2, "result": {"content": [{"type": "text", diff --git a/crates/memory/src/mcp_meta.rs b/crates/memory/src/mcp_meta.rs index 00a3f488..14187db5 100644 --- a/crates/memory/src/mcp_meta.rs +++ b/crates/memory/src/mcp_meta.rs @@ -73,13 +73,15 @@ pub fn request(id: u64, method: &str, mut params: Value, client: &str) -> Value } /// The headers a streamable HTTP POST of `request` carries: its version, its method, -/// and, for a method that names something, that name. +/// and, for a method that names something, that name. Only the name may go encoded +/// ([`encode`]); 2026-07-28 allows that for `Mcp-Name` and `Mcp-Param-*` alone, and a +/// server compares `Mcp-Method` with the body as sent. #[must_use] pub fn headers(request: &Value) -> Vec<(&'static str, String)> { let method = request["method"].as_str().unwrap_or_default(); let mut out = vec![ (VERSION_HEADER, MODERN.to_string()), - (METHOD_HEADER, encode(method)), + (METHOD_HEADER, method.to_string()), ]; if let Some(key) = named_by(method) { if let Some(name) = request["params"][key].as_str() { @@ -170,6 +172,15 @@ mod tests { assert_eq!(headers(&d).len(), 2); } + /// 2026-07-28 lets the sentinel stand for a value in `Mcp-Name` and `Mcp-Param-*` + /// only; `Mcp-Method` carries the method as it is, even one that looks like it. + #[test] + fn the_method_header_carries_the_method_as_it_is() { + let looks_encoded = "=?base64?dG9vbHMvbGlzdA==?="; + let r = request(1, looks_encoded, Value::Null, "t"); + assert_eq!(headers(&r)[1], (METHOD_HEADER, looks_encoded.to_string())); + } + #[test] fn a_value_unsafe_in_a_header_goes_as_base64_and_comes_back() { assert_eq!(encode("fetch"), "fetch"); diff --git a/crates/memory/src/tool_server.rs b/crates/memory/src/tool_server.rs index d20ea101..28bdc786 100644 --- a/crates/memory/src/tool_server.rs +++ b/crates/memory/src/tool_server.rs @@ -226,7 +226,10 @@ fn checked(method: &str, params: &Value, headers: &HeaderMap) -> Result<(), Refu /// Whether the method and name headers, where sent, say what the body says. Whoever /// sends them, a request whose headers name another method or tool than its body is /// refused: whatever routed on the headers acted on a different request than this one. -/// A client of the handshake may leave them out. +/// A client of the handshake may leave them out. The name is decoded before it is +/// compared; the method is not, since 2026-07-28 allows the base64 sentinel in +/// `Mcp-Name` and `Mcp-Param-*` only, and whatever routed on `Mcp-Method` read it as +/// sent. fn mirrored(method: &str, params: &Value, headers: &HeaderMap) -> Result<(), Refusal> { if let Some(m) = header(headers, METHOD_HEADER)? { if m != method { @@ -689,6 +692,22 @@ mod tests { assert_eq!(got["error"]["code"], -32601, "{got}"); } + /// 2026-07-28 lets the sentinel stand for a value in `Mcp-Name` and `Mcp-Param-*` + /// only, so a method header in it names another method than the body's. + #[tokio::test] + async fn a_method_header_is_compared_as_sent_and_never_decoded() { + let app = app(); + let (r, mut h) = modern("tools/list", Value::Null); + h.retain(|(k, _)| *k != mcp_meta::METHOD_HEADER); + h.push(( + mcp_meta::METHOD_HEADER, + "=?base64?dG9vbHMvbGlzdA==?=".into(), + )); + let (status, got) = post_to(&app, &h, &r).await; + assert_eq!(status, 400, "{got}"); + assert_eq!(got["error"]["code"], mcp_meta::HEADER_MISMATCH, "{got}"); + } + #[tokio::test] async fn a_client_that_opens_with_initialize_is_answered_as_before() { let app = app(); From c26100748af55b830d90be0e4d9fadb2c3b3878a Mon Sep 17 00:00:00 2001 From: Sander Striker Date: Mon, 2 Nov 2026 23:41:30 +0100 Subject: [PATCH 4/5] test: show that parse and request errors clear the cached revision A 200 reply with error -32700 or -32600 clears the cached revision, but no test showed it. Two cases now do. The hook client also refuses a header value that has a line break or another control byte, because the request is written as raw text and the method header is no longer encoded. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Bhj4Ck4y5iwZUsjpgQx8A2 --- crates/memory/src/hosted.rs | 8 ++++++++ crates/memory/src/mcp_http.rs | 17 +++++++++++++++++ 2 files changed, 25 insertions(+) diff --git a/crates/memory/src/hosted.rs b/crates/memory/src/hosted.rs index cee93380..9516fd15 100644 --- a/crates/memory/src/hosted.rs +++ b/crates/memory/src/hosted.rs @@ -649,6 +649,14 @@ mod tests { j, r#"{"jsonrpc":"2.0","id":1,"error":{"message":"no code"}}"#, ), + ( + j, + r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32700,"message":"parse error"}}"#, + ), + ( + j, + r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32600,"message":"invalid request"}}"#, + ), ]; for (n, (kind, body)) in unspoken.into_iter().enumerate() { *f.answer.lock().unwrap() = Some((200, kind, body)); diff --git a/crates/memory/src/mcp_http.rs b/crates/memory/src/mcp_http.rs index 1cd0cf84..d3b81662 100644 --- a/crates/memory/src/mcp_http.rs +++ b/crates/memory/src/mcp_http.rs @@ -61,6 +61,14 @@ fn dechunk(raw: &str) -> String { /// after it, so the end of the stream is the end of the answer. async fn post(url: &str, headers: &[(&str, String)], body: &str) -> Result { let (authority, path) = split(url)?; + // A header value is written into the request as it is, so a line break in one + // would start a header of its own. + if let Some((name, _)) = headers + .iter() + .find(|(_, value)| !value.bytes().all(|b| (0x20..0x7f).contains(&b))) + { + return Err(format!("{name}: the value has a byte a header cannot carry")); + } let mut s = TcpStream::connect(&authority) .await .map_err(|e| format!("{authority}: {e}"))?; @@ -182,6 +190,15 @@ mod tests { assert!(split("http:///mcp").is_err()); } + #[tokio::test] + async fn a_header_value_with_a_line_break_is_refused_before_anything_is_sent() { + let headers = [("mcp-method", "tools/call\r\nx-injected: 1".to_string())]; + let Err(refused) = post("http://127.0.0.1:1/mcp", &headers, "{}").await else { + panic!("a request with a broken header was sent"); + }; + assert!(refused.contains("cannot carry"), "{refused}"); + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn a_tool_call_reaches_the_server_and_a_refusal_comes_back_in_words() { let dir = tempfile::tempdir().unwrap(); From 8feb0cb25329f52713aae91aa658e1474f069f52 Mon Sep 17 00:00:00 2001 From: Sander Striker Date: Tue, 3 Nov 2026 00:11:28 +0100 Subject: [PATCH 5/5] style: format the header test Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Bhj4Ck4y5iwZUsjpgQx8A2 --- crates/memory/src/mcp_http.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/crates/memory/src/mcp_http.rs b/crates/memory/src/mcp_http.rs index d3b81662..658a57bf 100644 --- a/crates/memory/src/mcp_http.rs +++ b/crates/memory/src/mcp_http.rs @@ -67,7 +67,9 @@ async fn post(url: &str, headers: &[(&str, String)], body: &str) -> Result