OXYGENHUB
# 01320eefd505 spectra / crates / spectra-mcp / src / protocol.rs
6.8 KiB · Rust · last changed The 0.1 tools: brief, rules, agents, status, say, read, inbox, wait with the idle guard, ack, search, redact; every call on the record (#8) 2 days ago source blame history raw
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 
pub fn error(id: &Json, code: i64, message: &str) -> Json {
40 
    json!({ "jsonrpc": "2.0", "id": id, "error": { "code": code, "message": message } })
41 
}
42
43 
fn ok(id: &Json, version: &str, mut result: Json, meta: Json) -> Json {
44 
    if let Some(o) = result.as_object_mut() {
45 
        if version == STATELESS {
46 
            o.insert("resultType".into(), json!("complete"));
47 
        }
48 
        let mut m = meta.as_object().cloned().unwrap_or_default();
49 
        if version == STATELESS {
50 
            m.insert("io.modelcontextprotocol/serverInfo".into(), server_info());
51 
        }
52 
        if !m.is_empty() {
53 
            o.insert("_meta".into(), Json::Object(m));
54 
        }
55 
    }
56 
    json!({ "jsonrpc": "2.0", "id": id, "result": result })
57 
}
58
59 
fn server_info() -> Json {
60 
    json!({ "name": "spectra", "version": env!("CARGO_PKG_VERSION") })
61 
}
62
63 
fn capabilities() -> Json {
64 
    json!({ "tools": { "listChanged": false } })
65 
}
66
67 
/// Check the mirrored headers of a 2026-07-28 request against its body.
68 
fn check_headers(h: &Headers, req: &Json) -> Result<(), String> {
69 
    let method = req.get("method").and_then(Json::as_str).unwrap_or("");
70 
    if h.method != Some(method) {
71 
        return Err(format!("Mcp-Method header {:?} does not match the body's {method:?}", h.method));
72 
    }
73 
    let body_version = req.pointer("/params/_meta/io.modelcontextprotocol~1protocolVersion").and_then(Json::as_str);
74 
    if body_version.is_some_and(|v| Some(v) != h.version) {
75 
        return Err("MCP-Protocol-Version does not match the body's protocolVersion".into());
76 
    }
77 
    if matches!(method, "tools/call" | "resources/read" | "prompts/get") {
78 
        let name = req.pointer("/params/name").or_else(|| req.pointer("/params/uri")).and_then(Json::as_str);
79 
        if name.is_none() || h.name != name {
80 
            return Err(format!("Mcp-Name header {:?} does not match the body's {name:?}", h.name));
81 
        }
82 
    }
83 
    Ok(())
84 
}
85
86 
/// Answer one JSON-RPC message from an authenticated caller.
87 
pub async fn handle(ctx: &Ctx, h: &Headers<'_>, req: Json, meta: Json) -> Reply {
88 
    let _ = &meta;
89 
    let id = req.get("id").cloned().unwrap_or(Json::Null);
90 
    let Some(method) = req.get("method").and_then(Json::as_str).map(String::from) else {
91 
        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.
94 
    let version = h.version.unwrap_or("2025-03-26");
95 
    if !PROTOCOL_VERSIONS.contains(&version) {
96 
        let mut e = error(&id, UNSUPPORTED_VERSION, &format!("unsupported protocol version {version}"));
97 
        e["error"]["data"] = json!({ "supported": PROTOCOL_VERSIONS });
98 
        return Reply::Json(400, e);
99 
    }
100 
    if version == STATELESS {
101 
        if let Err(why) = check_headers(h, &req) {
102 
            return Reply::Json(400, error(&id, HEADER_MISMATCH, &why));
103 
        }
104 
    }
105 
    let is_notification = req.get("id").is_none();
106 
    match method.as_str() {
107 
        "initialize" => {
108 
            let asked = req.pointer("/params/protocolVersion").and_then(Json::as_str).unwrap_or("2025-03-26");
109 
            let chosen = if PROTOCOL_VERSIONS[1..].contains(&asked) { asked } else { PROTOCOL_VERSIONS[1] };
110 
            let r = json!({ "protocolVersion": chosen, "capabilities": capabilities(), "serverInfo": server_info(), "instructions": INSTRUCTIONS });
111 
            Reply::Json(200, ok(&id, chosen, r, meta))
112 
        }
113 
        "server/discover" => {
114 
            let r = json!({ "supportedVersions": PROTOCOL_VERSIONS, "capabilities": capabilities(), "serverInfo": server_info(), "instructions": INSTRUCTIONS });
115 
            Reply::Json(200, ok(&id, version, r, meta))
116 
        }
117 
        "ping" => Reply::Json(200, ok(&id, version, json!({}), meta)),
118 
        _ if is_notification => Reply::Accepted,
119 
        "tools/list" => {
120 
            let tools: Vec<Json> = tools::all().iter().map(|t| t.describe()).collect();
121 
            let mut r = json!({ "tools": tools });
122 
            if version == STATELESS {
123 
                r["ttlMs"] = json!(3_600_000);
124 
                r["cacheScope"] = json!("private");
125 
            }
126 
            Reply::Json(200, ok(&id, version, r, meta))
127 
        }
128 
        "tools/call" => {
129 
            let name = req.pointer("/params/name").and_then(Json::as_str).unwrap_or("");
130 
            let Some(tool) = tools::find(name) else {
131 
                return Reply::Json(200, error(&id, INVALID_PARAMS, &format!("no tool named {name}")));
132 
            };
133 
            let args = req.pointer("/params/arguments").cloned().unwrap_or_else(|| json!({}));
134 
            if tool.long {
135 
                return Reply::Stream(id, tool, args);
136 
            }
137 
            let out = tools::invoke(ctx, tool, args).await;
138 
            let meta = crate::http::meta(ctx).await;
139 
            Reply::Json(200, ok(&id, version, tool_result(&out), meta))
140 
        }
141 
        "subscriptions/listen" if version == STATELESS => Reply::Listen(id, spectra_core::ids::ulid()),
142 
        _ => Reply::Json(if version == STATELESS { 404 } else { 200 }, error(&id, METHOD_NOT_FOUND, &format!("no method {method}"))),
143 
    }
144 
}
145
146 
/// A tool's output as a `tools/call` result.
147 
pub fn tool_result(out: &Output) -> Json {
148 
    let mut r = json!({ "content": [{ "type": "text", "text": out.text }], "isError": out.is_error });
149 
    if let Some(s) = &out.structured {
150 
        r["structuredContent"] = s.clone();
151 
    }
152 
    r
153 
}
154
155 
/// The final message of a streamed tool call.
156 
pub fn stream_result(id: &Json, version: &str, out: &Output, meta: Json) -> Json {
157 
    ok(id, version, tool_result(out), meta)
158 
}