mcp.rs 20.6%

14 of 68 lines run, 3 of 16 functions · run #22 on main · run missed

//! `spectra mcp --agent NAME`: the stdio shim (spec §15.1). A client starts
//! it as an ordinary stdio MCP server; it signs every request with the
//! agent's key from the agent's own home and forwards it to the daemon over
//! the Unix socket, so a client's configuration is one line with nothing
//! secret in it. Nothing is written to stdout but protocol.

use super::Ctx;
use anyhow::{bail, Context, Result};
use serde_json::{json, Value as Json};
use spectra_core::keys::KeyPair;
use spectra_core::ledger::Ledger;
use spectra_mcp::client::{rpc_result, Client};
use std::path::PathBuf;
use std::sync::Arc;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::sync::Mutex;

/// The revisions the shim speaks to its client (spec §15.1); 2026-07-28
/// requests are passed through as they are.
pub const STDIO_VERSIONS: &[&str] = &["2025-11-25", "2025-06-18", "2025-03-26"];

pub async fn run(ctx: &Ctx, agent: &str, home: Option<PathBuf>) -> Result<i32> {
    let home = match home {
        Some(h) => h,
        None if ctx.remote()?.is_some() => {
            bail!("with Spectra on another machine, name the agent's home: spectra mcp --agent {agent} --home DIR")
        }
        None => {
            let ledger = Ledger::open_read_only(&ctx.db()?).await?;
            let h: Option<String> =
                sqlx::query_scalar("SELECT home FROM agents WHERE name = ?").bind(agent).fetch_optional(ledger.pool()).await?;
            PathBuf::from(h.with_context(|| format!("no agent named {agent} (spectra agent list)"))?)
        }
    };
    let key = KeyPair::load_home(&home).with_context(|| format!("reading {agent}'s key in {}", home.display()))?;
    if key.identity.name != agent {
        bail!("{} holds {}'s key, not {agent}'s", home.display(), key.identity.name);
    }
    let client = Arc::new(Client::at(ctx.endpoint()?, Some(key)));
    let version = Arc::new(Mutex::new(STDIO_VERSIONS[0].to_string()));
    let out = Arc::new(Mutex::new(tokio::io::stdout()));
    let mut lines = BufReader::new(tokio::io::stdin()).lines();
    let mut tasks = tokio::task::JoinSet::new();
    while let Some(line) = lines.next_line().await? {
        if line.trim().is_empty() {
            continue;
        }
        let (client, version, out) = (client.clone(), version.clone(), out.clone());
        // Each request on its own: a `wait` must not hold up the others.
        tasks.spawn(async move {
            if let Some(reply) = forward(&client, &version, &line).await {
                let mut o = out.lock().await;
                let _ = o.write_all(format!("{reply}\n").as_bytes()).await;
                let _ = o.flush().await;
            }
        });
    }
    while tasks.join_next().await.is_some() {}
    Ok(0)
}

/// Forward one message; the reply to write, if it was a request.
async fn forward(client: &Client, version: &Mutex<String>, line: &str) -> Option<Json> {
    let req: Json = match serde_json::from_str(line) {
        Ok(v) => v,
        Err(e) => return Some(json!({ "jsonrpc": "2.0", "id": null, "error": { "code": -32700, "message": format!("not JSON: {e}") } })),
    };
    let id = req.get("id").cloned();
    let method = req.get("method").and_then(Json::as_str).unwrap_or("");
    let asked_2026 = method == "server/discover"
        || req.pointer("/params/_meta/io.modelcontextprotocol~1protocolVersion").and_then(Json::as_str) == Some("2026-07-28");
    let v = if asked_2026 {
        "2026-07-28".to_string()
    } else if method == "initialize" {
        let asked = req.pointer("/params/protocolVersion").and_then(Json::as_str).unwrap_or(STDIO_VERSIONS[2]);
        let chosen = if STDIO_VERSIONS.contains(&asked) { asked } else { STDIO_VERSIONS[0] };
        *version.lock().await = chosen.to_string();
        chosen.to_string()
    } else {
        version.lock().await.clone()
    };
    let r = match client.mcp(&v, &req).await {
        Ok(r) => r,
        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?") } })),
    };
    let id = id?;
    if r.status == 202 {
        return None;
    }
    match rpc_result(&r) {
        Ok(v) => Some(v),
        Err(e) => Some(
            json!({ "jsonrpc": "2.0", "id": id, "error": { "code": -32000, "message": format!("Spectra answered {}: {e:#}", r.status) } }),
        ),
    }
}

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	2×	pub async fn run(ctx: &Ctx, agent: &str, home: Option<PathBuf>) -> Result<i32> {
23	1×	let home = match home {
24	1×	Some(h) => h,
25	1×	None if ctx.remote()?.is_some() => {
26	0×	bail!("with Spectra on another machine, name the agent's home: spectra mcp --agent {agent} --home DIR")
27		}
28		None => {
29	1×	let ledger = Ledger::open_read_only(&ctx.db()?).await?;
30	1×	let h: Option<String> =
31	1×	sqlx::query_scalar("SELECT home FROM agents WHERE name = ?").bind(agent).fetch_optional(ledger.pool()).await?;
32	1×	PathBuf::from(h.with_context(\|\| format!("no agent named {agent} (spectra agent list)"))?)
33		}
34		};
35	1×	let key = KeyPair::load_home(&home).with_context(\|\| format!("reading {agent}'s key in {}", home.display()))?;
36	1×	if key.identity.name != agent {
37	1×	bail!("{} holds {}'s key, not {agent}'s", home.display(), key.identity.name);
38	0×	}
39	0×	let client = Arc::new(Client::at(ctx.endpoint()?, Some(key)));
40	0×	let version = Arc::new(Mutex::new(STDIO_VERSIONS[0].to_string()));
41	0×	let out = Arc::new(Mutex::new(tokio::io::stdout()));
42	0×	let mut lines = BufReader::new(tokio::io::stdin()).lines();
43	0×	let mut tasks = tokio::task::JoinSet::new();
44	0×	while let Some(line) = lines.next_line().await? {
45	0×	if line.trim().is_empty() {
46	0×	continue;
47	0×	}
48	0×	let (client, version, out) = (client.clone(), version.clone(), out.clone());
49		// Each request on its own: a `wait` must not hold up the others.
50	0×	tasks.spawn(async move {
51	0×	if let Some(reply) = forward(&client, &version, &line).await {
52	0×	let mut o = out.lock().await;
53	0×	let _ = o.write_all(format!("{reply}\n").as_bytes()).await;
54	0×	let _ = o.flush().await;
55	0×	}
56	0×	});
57		}
58	0×	while tasks.join_next().await.is_some() {}
59	0×	Ok(0)
60	2×	}
61
62		/// Forward one message; the reply to write, if it was a request.
63	0×	async fn forward(client: &Client, version: &Mutex<String>, line: &str) -> Option<Json> {
64	0×	let req: Json = match serde_json::from_str(line) {
65	0×	Ok(v) => v,
66	0×	Err(e) => return Some(json!({ "jsonrpc": "2.0", "id": null, "error": { "code": -32700, "message": format!("not JSON: {e}") } })),
67		};
68	0×	let id = req.get("id").cloned();
69	0×	let method = req.get("method").and_then(Json::as_str).unwrap_or("");
70	0×	let asked_2026 = method == "server/discover"
71	0×	\|\| req.pointer("/params/_meta/io.modelcontextprotocol~1protocolVersion").and_then(Json::as_str) == Some("2026-07-28");
72	0×	let v = if asked_2026 {
73	0×	"2026-07-28".to_string()
74	0×	} else if method == "initialize" {
75	0×	let asked = req.pointer("/params/protocolVersion").and_then(Json::as_str).unwrap_or(STDIO_VERSIONS[2]);
76	0×	let chosen = if STDIO_VERSIONS.contains(&asked) { asked } else { STDIO_VERSIONS[0] };
77	0×	*version.lock().await = chosen.to_string();
78	0×	chosen.to_string()
79		} else {
80	0×	version.lock().await.clone()
81		};
82	0×	let r = match client.mcp(&v, &req).await {
83	0×	Ok(r) => r,
84	0×	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?") } })),
85		};
86	0×	let id = id?;
87	0×	if r.status == 202 {
88	0×	return None;
89	0×	}
90	0×	match rpc_result(&r) {
91	0×	Ok(v) => Some(v),
92	0×	Err(e) => Some(
93	0×	json!({ "jsonrpc": "2.0", "id": id, "error": { "code": -32000, "message": format!("Spectra answered {}: {e:#}", r.status) } }),
94	0×	),
95		}
96	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 / src/cli/mcp.rs 20.6%

Keyboard