OXYGENHUB
# b576d05cf862 spectra / src / cli / mcp.rs
4.2 KiB · Rust · last changed The stdio shim: spectra mcp --agent NAME signs with the agent's key and forwards to the daemon (#9) 2 days ago source blame history raw
1
//! `spectra mcp --agent NAME`: the stdio shim (spec §15.1). A client starts
2
//! it as an ordinary stdio MCP server; it signs every request with the
3
//! agent's key from the agent's own home and forwards it to the daemon over
4
//! the Unix socket, so a client's configuration is one line with nothing
5
//! secret in it. Nothing is written to stdout but protocol.
6
7
use super::Ctx;
8
use anyhow::{bail, Context, Result};
9
use serde_json::{json, Value as Json};
10
use spectra_core::keys::KeyPair;
11
use spectra_core::ledger::Ledger;
12
use spectra_mcp::client::{rpc_result, Client};
13
use std::path::PathBuf;
14
use std::sync::Arc;
15
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
16
use tokio::sync::Mutex;
17
18
/// The revisions the shim speaks to its client (spec §15.1); 2026-07-28
19
/// requests are passed through as they are.
20
pub const STDIO_VERSIONS: &[&str] = &["2025-11-25", "2025-06-18", "2025-03-26"];
21
22
pub async fn run(ctx: &Ctx, agent: &str, home: Option<PathBuf>) -> Result<i32> {
23
    let home = match home {
24
        Some(h) => h,
25
        None => {
26
            let ledger = Ledger::open_read_only(&ctx.db()?).await?;
27
            let h: Option<String> =
28
                sqlx::query_scalar("SELECT home FROM agents WHERE name = ?").bind(agent).fetch_optional(ledger.pool()).await?;
29
            PathBuf::from(h.with_context(|| format!("no agent named {agent} (spectra agent list)"))?)
30
        }
31
    };
32
    let key = KeyPair::load_home(&home).with_context(|| format!("reading {agent}'s key in {}", home.display()))?;
33
    if key.identity.name != agent {
34
        bail!("{} holds {}'s key, not {agent}'s", home.display(), key.identity.name);
35
    }
36
    let client = Arc::new(Client::new(&ctx.socket(), Some(key)));
37
    let version = Arc::new(Mutex::new(STDIO_VERSIONS[0].to_string()));
38
    let out = Arc::new(Mutex::new(tokio::io::stdout()));
39
    let mut lines = BufReader::new(tokio::io::stdin()).lines();
40
    let mut tasks = tokio::task::JoinSet::new();
41
    while let Some(line) = lines.next_line().await? {
42
        if line.trim().is_empty() {
43
            continue;
44
        }
45
        let (client, version, out) = (client.clone(), version.clone(), out.clone());
46
        // Each request on its own: a `wait` must not hold up the others.
47
        tasks.spawn(async move {
48
            if let Some(reply) = forward(&client, &version, &line).await {
49
                let mut o = out.lock().await;
50
                let _ = o.write_all(format!("{reply}\n").as_bytes()).await;
51
                let _ = o.flush().await;
52
            }
53
        });
54
    }
55
    while tasks.join_next().await.is_some() {}
56
    Ok(0)
57
}
58
59
/// Forward one message; the reply to write, if it was a request.
60
async fn forward(client: &Client, version: &Mutex<String>, line: &str) -> Option<Json> {
61
    let req: Json = match serde_json::from_str(line) {
62
        Ok(v) => v,
63
        Err(e) => return Some(json!({ "jsonrpc": "2.0", "id": null, "error": { "code": -32700, "message": format!("not JSON: {e}") } })),
64
    };
65
    let id = req.get("id").cloned();
66
    let method = req.get("method").and_then(Json::as_str).unwrap_or("");
67
    let asked_2026 = method == "server/discover"
68
        || req.pointer("/params/_meta/io.modelcontextprotocol~1protocolVersion").and_then(Json::as_str) == Some("2026-07-28");
69
    let v = if asked_2026 {
70
        "2026-07-28".to_string()
71
    } else if method == "initialize" {
72
        let asked = req.pointer("/params/protocolVersion").and_then(Json::as_str).unwrap_or(STDIO_VERSIONS[2]);
73
        let chosen = if STDIO_VERSIONS.contains(&asked) { asked } else { STDIO_VERSIONS[0] };
74
        *version.lock().await = chosen.to_string();
75
        chosen.to_string()
76
    } else {
77
        version.lock().await.clone()
78
    };
79
    let r = match client.mcp(&v, &req).await {
80
        Ok(r) => r,
81
        Err(e) => return id.map(|id| json!({ "jsonrpc": "2.0", "id": id, "error": { "code": -32000, "message": format!("Spectra is not reachable: {e:#}. Is `spectra serve` running?") } })),
82
    };
83
    let id = id?;
84
    if r.status == 202 {
85
        return None;
86
    }
87
    match rpc_result(&r) {
88
        Ok(v) => Some(v),
89
        Err(e) => Some(
90
            json!({ "jsonrpc": "2.0", "id": id, "error": { "code": -32000, "message": format!("Spectra answered {}: {e:#}", r.status) } }),
91
        ),
92
    }
93
}