OXYGENHUB

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

97 of 106 lines run, 11 of 26 functions · run #15 on main · run missed

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
397×
pub fn error(id: &Json, code: i64, message: &str) -> Json {
407×
    json!({ "jsonrpc": "2.0", "id": id, "error": { "code": code, "message": message } })
417×
}
42
4336×
fn ok(id: &Json, version: &str, mut result: Json, meta: Json) -> Json {
4436×
    if let Some(o) = result.as_object_mut() {
4536×
        if version == STATELESS {
4626×
            o.insert("resultType".into(), json!("complete"));
4726×
        }
4836×
        let mut m = meta.as_object().cloned().unwrap_or_default();
4936×
        if version == STATELESS {
5026×
            m.insert("io.modelcontextprotocol/serverInfo".into(), server_info());
5126×
        }
5236×
        if !m.is_empty() {
5336×
            o.insert("_meta".into(), Json::Object(m));
5436×
        }
550×
    }
5636×
    json!({ "jsonrpc": "2.0", "id": id, "result": result })
5736×
}
58
5929×
fn server_info() -> Json {
6029×
    json!({ "name": "spectra", "version": env!("CARGO_PKG_VERSION") })
6129×
}
62
633×
fn capabilities() -> Json {
643×
    json!({ "tools": { "listChanged": false } })
653×
}
66
67
/// Check the mirrored headers of a 2026-07-28 request against its body.
6827×
fn check_headers(h: &Headers, req: &Json) -> Result<(), String> {
6927×
    let method = req.get("method").and_then(Json::as_str).unwrap_or("");
7027×
    if h.method != Some(method) {
711×
        return Err(format!("Mcp-Method header {:?} does not match the body's {method:?}", h.method));
7226×
    }
7326×
    let body_version = req.pointer("/params/_meta/io.modelcontextprotocol~1protocolVersion").and_then(Json::as_str);
7426×
    if body_version.is_some_and(|v| Some(v) != h.version) {
750×
        return Err("MCP-Protocol-Version does not match the body's protocolVersion".into());
7626×
    }
7726×
    if matches!(method, "tools/call" | "resources/read" | "prompts/get") {
7823×
        let name = req.pointer("/params/name").or_else(|| req.pointer("/params/uri")).and_then(Json::as_str);
7923×
        if name.is_none() || h.name != name {
800×
            return Err(format!("Mcp-Name header {:?} does not match the body's {name:?}", h.name));
8123×
        }
823×
    }
8326×
    Ok(())
8427×
}
85
86
/// Answer one JSON-RPC message from an authenticated caller.
8740×
pub async fn handle(ctx: &Ctx, h: &Headers<'_>, req: Json, meta: Json) -> Reply {
8840×
    let _ = &meta;
8940×
    let id = req.get("id").cloned().unwrap_or(Json::Null);
9040×
    let Some(method) = req.get("method").and_then(Json::as_str).map(String::from) else {
910×
        return Reply::Json(400, error(&id, INVALID_REQUEST, "a JSON-RPC request needs a method"));
92
    };
93
    // The revision: the header, else the legacy default of the transport.
9440×
    let version = h.version.unwrap_or("2025-03-26");
9540×
    if !PROTOCOL_VERSIONS.contains(&version) {
961×
        let mut e = error(&id, UNSUPPORTED_VERSION, &format!("unsupported protocol version {version}"));
971×
        e["error"]["data"] = json!({ "supported": PROTOCOL_VERSIONS });
981×
        return Reply::Json(400, e);
9939×
    }
10039×
    if version == STATELESS {
10127×
        if let Err(why) = check_headers(h, &req) {
1021×
            return Reply::Json(400, error(&id, HEADER_MISMATCH, &why));
10326×
        }
10412×
    }
10538×
    let is_notification = req.get("id").is_none();
10638×
    match method.as_str() {
10738×
        "initialize" => {
1082×
            let asked = req.pointer("/params/protocolVersion").and_then(Json::as_str).unwrap_or("2025-03-26");
1092×
            let chosen = if PROTOCOL_VERSIONS[1..].contains(&asked) { asked } else { PROTOCOL_VERSIONS[1] };
1102×
            let r = json!({ "protocolVersion": chosen, "capabilities": capabilities(), "serverInfo": server_info(), "instructions": INSTRUCTIONS });
1112×
            Reply::Json(200, ok(&id, chosen, r, meta))
112
        }
11336×
        "server/discover" => {
1141×
            let r = json!({ "supportedVersions": PROTOCOL_VERSIONS, "capabilities": capabilities(), "serverInfo": server_info(), "instructions": INSTRUCTIONS });
1151×
            Reply::Json(200, ok(&id, version, r, meta))
116
        }
11735×
        "ping" => Reply::Json(200, ok(&id, version, json!({}), meta)),
1182×
        _ if is_notification => Reply::Accepted,
11932×
        "tools/list" => {
12044×
            let tools: Vec<Json> = tools::all().iter().map(|t| t.describe()).collect();
1214×
            let mut r = json!({ "tools": tools });
1224×
            if version == STATELESS {
1232×
                r["ttlMs"] = json!(3_600_000);
1242×
                r["cacheScope"] = json!("private");
1252×
            }
1264×
            Reply::Json(200, ok(&id, version, r, meta))
127
        }
12828×
        "tools/call" => {
12928×
            let name = req.pointer("/params/name").and_then(Json::as_str).unwrap_or("");
13028×
            let Some(tool) = tools::find(name) else {
1310×
                return Reply::Json(200, error(&id, INVALID_PARAMS, &format!("no tool named {name}")));
132
            };
13328×
            let args = req.pointer("/params/arguments").cloned().unwrap_or_else(|| json!({}));
13428×
            if tool.long {
1356×
                return Reply::Stream(id, tool, args);
13622×
            }
13722×
            let out = tools::invoke(ctx, tool, args).await;
13822×
            let meta = crate::http::meta(ctx).await;
13922×
            Reply::Json(200, ok(&id, version, tool_result(&out), meta))
140
        }
1410×
        "subscriptions/listen" if version == STATELESS => Reply::Listen(id, spectra_core::ids::ulid()),
1420×
        _ => Reply::Json(if version == STATELESS { 404 } else { 200 }, error(&id, METHOD_NOT_FOUND, &format!("no method {method}"))),
143
    }
14440×
}
145
146
/// A tool's output as a `tools/call` result.
14728×
pub fn tool_result(out: &Output) -> Json {
14828×
    let mut r = json!({ "content": [{ "type": "text", "text": out.text }], "isError": out.is_error });
14928×
    if let Some(s) = &out.structured {
15025×
        r["structuredContent"] = s.clone();
15125×
    }
15228×
    r
15328×
}
154
155
/// The final message of a streamed tool call.
1566×
pub fn stream_result(id: &Json, version: &str, out: &Output, meta: Json) -> Json {
1576×
    ok(id, version, tool_result(out), meta)
1586×
}