protocol.rs 87.5%

91 of 104 lines run, 10 of 26 functions · run #8 on main · run missed

//! The MCP protocol (spec §15.1): revision 2026-07-28, stateless, with its
//! mirrored headers checked against the body; and 2025-11-25, 2025-06-18
//! and 2025-03-26, which begin with `initialize`. No sessions are minted;
//! Spectra's own session (spec §8.2) rides in `_meta`.

use crate::tools::{self, Ctx, Output};
use crate::{INSTRUCTIONS, PROTOCOL_VERSIONS, STATELESS};
use serde_json::{json, Value as Json};

/// JSON-RPC and MCP error codes.
pub const PARSE_ERROR: i64 = -32700;
pub const INVALID_REQUEST: i64 = -32600;
pub const METHOD_NOT_FOUND: i64 = -32601;
pub const INVALID_PARAMS: i64 = -32602;
pub const HEADER_MISMATCH: i64 = -32020;
pub const UNSUPPORTED_VERSION: i64 = -32022;
/// Implementation-defined: the request did not authenticate (spec §8.5).
pub const UNAUTHENTICATED: i64 = -32010;

/// How a request is answered.
pub enum Reply {
    /// A JSON body with this HTTP status.
    Json(u16, Json),
    /// 202 and no body: a notification.
    Accepted,
    /// A tool that takes long: run it, answering as SSE with keep-alives.
    Stream(Json, &'static tools::Tool, Json),
    /// `subscriptions/listen`: an SSE stream that stays open.
    Listen(Json, String),
}

/// The revision a request speaks, and the headers it mirrors.
pub struct Headers<'a> {
    pub version: Option<&'a str>,
    pub method: Option<&'a str>,
    pub name: Option<&'a str>,
}

pub fn error(id: &Json, code: i64, message: &str) -> Json {
    json!({ "jsonrpc": "2.0", "id": id, "error": { "code": code, "message": message } })
}

fn ok(id: &Json, version: &str, mut result: Json, meta: Json) -> Json {
    if let Some(o) = result.as_object_mut() {
        if version == STATELESS {
            o.insert("resultType".into(), json!("complete"));
        }
        let mut m = meta.as_object().cloned().unwrap_or_default();
        if version == STATELESS {
            m.insert("io.modelcontextprotocol/serverInfo".into(), server_info());
        }
        if !m.is_empty() {
            o.insert("_meta".into(), Json::Object(m));
        }
    }
    json!({ "jsonrpc": "2.0", "id": id, "result": result })
}

fn server_info() -> Json {
    json!({ "name": "spectra", "version": env!("CARGO_PKG_VERSION") })
}

fn capabilities() -> Json {
    json!({ "tools": { "listChanged": false } })
}

/// Check the mirrored headers of a 2026-07-28 request against its body.
fn check_headers(h: &Headers, req: &Json) -> Result<(), String> {
    let method = req.get("method").and_then(Json::as_str).unwrap_or("");
    if h.method != Some(method) {
        return Err(format!("Mcp-Method header {:?} does not match the body's {method:?}", h.method));
    }
    let body_version = req.pointer("/params/_meta/io.modelcontextprotocol~1protocolVersion").and_then(Json::as_str);
    if body_version.is_some_and(|v| Some(v) != h.version) {
        return Err("MCP-Protocol-Version does not match the body's protocolVersion".into());
    }
    if matches!(method, "tools/call" | "resources/read" | "prompts/get") {
        let name = req.pointer("/params/name").or_else(|| req.pointer("/params/uri")).and_then(Json::as_str);
        if name.is_none() || h.name != name {
            return Err(format!("Mcp-Name header {:?} does not match the body's {name:?}", h.name));
        }
    }
    Ok(())
}

/// Answer one JSON-RPC message from an authenticated caller.
pub async fn handle(ctx: &Ctx, h: &Headers<'_>, req: Json, meta: Json) -> Reply {
    let id = req.get("id").cloned().unwrap_or(Json::Null);
    let Some(method) = req.get("method").and_then(Json::as_str).map(String::from) else {
        return Reply::Json(400, error(&id, INVALID_REQUEST, "a JSON-RPC request needs a method"));
    };
    // The revision: the header, else the legacy default of the transport.
    let version = h.version.unwrap_or("2025-03-26");
    if !PROTOCOL_VERSIONS.contains(&version) {
        let mut e = error(&id, UNSUPPORTED_VERSION, &format!("unsupported protocol version {version}"));
        e["error"]["data"] = json!({ "supported": PROTOCOL_VERSIONS });
        return Reply::Json(400, e);
    }
    if version == STATELESS {
        if let Err(why) = check_headers(h, &req) {
            return Reply::Json(400, error(&id, HEADER_MISMATCH, &why));
        }
    }
    let is_notification = req.get("id").is_none();
    match method.as_str() {
        "initialize" => {
            let asked = req.pointer("/params/protocolVersion").and_then(Json::as_str).unwrap_or("2025-03-26");
            let chosen = if PROTOCOL_VERSIONS[1..].contains(&asked) { asked } else { PROTOCOL_VERSIONS[1] };
            let r = json!({ "protocolVersion": chosen, "capabilities": capabilities(), "serverInfo": server_info(), "instructions": INSTRUCTIONS });
            Reply::Json(200, ok(&id, chosen, r, meta))
        }
        "server/discover" => {
            let r = json!({ "supportedVersions": PROTOCOL_VERSIONS, "capabilities": capabilities(), "serverInfo": server_info(), "instructions": INSTRUCTIONS });
            Reply::Json(200, ok(&id, version, r, meta))
        }
        "ping" => Reply::Json(200, ok(&id, version, json!({}), meta)),
        _ if is_notification => Reply::Accepted,
        "tools/list" => {
            let tools: Vec<Json> = tools::all().iter().map(|t| t.describe()).collect();
            let mut r = json!({ "tools": tools });
            if version == STATELESS {
                r["ttlMs"] = json!(3_600_000);
                r["cacheScope"] = json!("private");
            }
            Reply::Json(200, ok(&id, version, r, meta))
        }
        "tools/call" => {
            let name = req.pointer("/params/name").and_then(Json::as_str).unwrap_or("");
            let Some(tool) = tools::find(name) else {
                return Reply::Json(200, error(&id, INVALID_PARAMS, &format!("no tool named {name}")));
            };
            let args = req.pointer("/params/arguments").cloned().unwrap_or_else(|| json!({}));
            if tool.long {
                return Reply::Stream(id, tool, args);
            }
            let out = (tool.run)(ctx, args).await;
            Reply::Json(200, ok(&id, version, tool_result(&out), meta))
        }
        "subscriptions/listen" if version == STATELESS => Reply::Listen(id, spectra_core::ids::ulid()),
        _ => Reply::Json(if version == STATELESS { 404 } else { 200 }, error(&id, METHOD_NOT_FOUND, &format!("no method {method}"))),
    }
}

/// A tool's output as a `tools/call` result.
pub fn tool_result(out: &Output) -> Json {
    let mut r = json!({ "content": [{ "type": "text", "text": out.text }], "isError": out.is_error });
    if let Some(s) = &out.structured {
        r["structuredContent"] = s.clone();
    }
    r
}

/// The final message of a streamed tool call.
pub fn stream_result(id: &Json, version: &str, out: &Output, meta: Json) -> Json {
    ok(id, version, tool_result(out), meta)
}

1		//! The MCP protocol (spec §15.1): revision 2026-07-28, stateless, with its
2		//! mirrored headers checked against the body; and 2025-11-25, 2025-06-18
3		//! and 2025-03-26, which begin with `initialize`. No sessions are minted;
4		//! Spectra's own session (spec §8.2) rides in `_meta`.
5
6		use crate::tools::{self, Ctx, Output};
7		use crate::{INSTRUCTIONS, PROTOCOL_VERSIONS, STATELESS};
8		use serde_json::{json, Value as Json};
9
10		/// JSON-RPC and MCP error codes.
11		pub const PARSE_ERROR: i64 = -32700;
12		pub const INVALID_REQUEST: i64 = -32600;
13		pub const METHOD_NOT_FOUND: i64 = -32601;
14		pub const INVALID_PARAMS: i64 = -32602;
15		pub const HEADER_MISMATCH: i64 = -32020;
16		pub const UNSUPPORTED_VERSION: i64 = -32022;
17		/// Implementation-defined: the request did not authenticate (spec §8.5).
18		pub const UNAUTHENTICATED: i64 = -32010;
19
20		/// How a request is answered.
21		pub enum Reply {
22		/// A JSON body with this HTTP status.
23		Json(u16, Json),
24		/// 202 and no body: a notification.
25		Accepted,
26		/// A tool that takes long: run it, answering as SSE with keep-alives.
27		Stream(Json, &'static tools::Tool, Json),
28		/// `subscriptions/listen`: an SSE stream that stays open.
29		Listen(Json, String),
30		}
31
32		/// The revision a request speaks, and the headers it mirrors.
33		pub struct Headers<'a> {
34		pub version: Option<&'a str>,
35		pub method: Option<&'a str>,
36		pub name: Option<&'a str>,
37		}
38
39	6×	pub fn error(id: &Json, code: i64, message: &str) -> Json {
40	6×	json!({ "jsonrpc": "2.0", "id": id, "error": { "code": code, "message": message } })
41	6×	}
42
43	5×	fn ok(id: &Json, version: &str, mut result: Json, meta: Json) -> Json {
44	5×	if let Some(o) = result.as_object_mut() {
45	5×	if version == STATELESS {
46	3×	o.insert("resultType".into(), json!("complete"));
47	3×	}
48	5×	let mut m = meta.as_object().cloned().unwrap_or_default();
49	5×	if version == STATELESS {
50	3×	m.insert("io.modelcontextprotocol/serverInfo".into(), server_info());
51	3×	}
52	5×	if !m.is_empty() {
53	5×	o.insert("_meta".into(), Json::Object(m));
54	5×	}
55	0×	}
56	5×	json!({ "jsonrpc": "2.0", "id": id, "result": result })
57	5×	}
58
59	5×	fn server_info() -> Json {
60	5×	json!({ "name": "spectra", "version": env!("CARGO_PKG_VERSION") })
61	5×	}
62
63	2×	fn capabilities() -> Json {
64	2×	json!({ "tools": { "listChanged": false } })
65	2×	}
66
67		/// Check the mirrored headers of a 2026-07-28 request against its body.
68	4×	fn check_headers(h: &Headers, req: &Json) -> Result<(), String> {
69	4×	let method = req.get("method").and_then(Json::as_str).unwrap_or("");
70	4×	if h.method != Some(method) {
71	1×	return Err(format!("Mcp-Method header {:?} does not match the body's {method:?}", h.method));
72	3×	}
73	3×	let body_version = req.pointer("/params/_meta/io.modelcontextprotocol~1protocolVersion").and_then(Json::as_str);
74	3×	if body_version.is_some_and(\|v\| Some(v) != h.version) {
75	0×	return Err("MCP-Protocol-Version does not match the body's protocolVersion".into());
76	3×	}
77	3×	if matches!(method, "tools/call" \| "resources/read" \| "prompts/get") {
78	1×	let name = req.pointer("/params/name").or_else(\|\| req.pointer("/params/uri")).and_then(Json::as_str);
79	1×	if name.is_none() \|\| h.name != name {
80	0×	return Err(format!("Mcp-Name header {:?} does not match the body's {name:?}", h.name));
81	1×	}
82	2×	}
83	3×	Ok(())
84	4×	}
85
86		/// Answer one JSON-RPC message from an authenticated caller.
87	8×	pub async fn handle(ctx: &Ctx, h: &Headers<'_>, req: Json, meta: Json) -> Reply {
88	8×	let id = req.get("id").cloned().unwrap_or(Json::Null);
89	8×	let Some(method) = req.get("method").and_then(Json::as_str).map(String::from) else {
90	0×	return Reply::Json(400, error(&id, INVALID_REQUEST, "a JSON-RPC request needs a method"));
91		};
92		// The revision: the header, else the legacy default of the transport.
93	8×	let version = h.version.unwrap_or("2025-03-26");
94	8×	if !PROTOCOL_VERSIONS.contains(&version) {
95	1×	let mut e = error(&id, UNSUPPORTED_VERSION, &format!("unsupported protocol version {version}"));
96	1×	e["error"]["data"] = json!({ "supported": PROTOCOL_VERSIONS });
97	1×	return Reply::Json(400, e);
98	7×	}
99	7×	if version == STATELESS {
100	4×	if let Err(why) = check_headers(h, &req) {
101	1×	return Reply::Json(400, error(&id, HEADER_MISMATCH, &why));
102	3×	}
103	3×	}
104	6×	let is_notification = req.get("id").is_none();
105	6×	match method.as_str() {
106	6×	"initialize" => {
107	1×	let asked = req.pointer("/params/protocolVersion").and_then(Json::as_str).unwrap_or("2025-03-26");
108	1×	let chosen = if PROTOCOL_VERSIONS[1..].contains(&asked) { asked } else { PROTOCOL_VERSIONS[1] };
109	1×	let r = json!({ "protocolVersion": chosen, "capabilities": capabilities(), "serverInfo": server_info(), "instructions": INSTRUCTIONS });
110	1×	Reply::Json(200, ok(&id, chosen, r, meta))
111		}
112	5×	"server/discover" => {
113	1×	let r = json!({ "supportedVersions": PROTOCOL_VERSIONS, "capabilities": capabilities(), "serverInfo": server_info(), "instructions": INSTRUCTIONS });
114	1×	Reply::Json(200, ok(&id, version, r, meta))
115		}
116	4×	"ping" => Reply::Json(200, ok(&id, version, json!({}), meta)),
117	1×	_ if is_notification => Reply::Accepted,
118	3×	"tools/list" => {
119	2×	let tools: Vec<Json> = tools::all().iter().map(\|t\| t.describe()).collect();
120	2×	let mut r = json!({ "tools": tools });
121	2×	if version == STATELESS {
122	1×	r["ttlMs"] = json!(3_600_000);
123	1×	r["cacheScope"] = json!("private");
124	1×	}
125	2×	Reply::Json(200, ok(&id, version, r, meta))
126		}
127	1×	"tools/call" => {
128	1×	let name = req.pointer("/params/name").and_then(Json::as_str).unwrap_or("");
129	1×	let Some(tool) = tools::find(name) else {
130	0×	return Reply::Json(200, error(&id, INVALID_PARAMS, &format!("no tool named {name}")));
131		};
132	1×	let args = req.pointer("/params/arguments").cloned().unwrap_or_else(\|\| json!({}));
133	1×	if tool.long {
134	0×	return Reply::Stream(id, tool, args);
135	1×	}
136	1×	let out = (tool.run)(ctx, args).await;
137	1×	Reply::Json(200, ok(&id, version, tool_result(&out), meta))
138		}
139	0×	"subscriptions/listen" if version == STATELESS => Reply::Listen(id, spectra_core::ids::ulid()),
140	0×	_ => Reply::Json(if version == STATELESS { 404 } else { 200 }, error(&id, METHOD_NOT_FOUND, &format!("no method {method}"))),
141		}
142	8×	}
143
144		/// A tool's output as a `tools/call` result.
145	1×	pub fn tool_result(out: &Output) -> Json {
146	1×	let mut r = json!({ "content": [{ "type": "text", "text": out.text }], "isError": out.is_error });
147	1×	if let Some(s) = &out.structured {
148	1×	r["structuredContent"] = s.clone();
149	1×	}
150	1×	r
151	1×	}
152
153		/// The final message of a streamed tool call.
154	0×	pub fn stream_result(id: &Json, version: &str, out: &Output, meta: Json) -> Json {
155	0×	ok(id, version, tool_result(out), meta)
156	0×	}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

Coverage / crates/spectra-mcp/src/protocol.rs 87.5%

Keyboard