The stdio shim: spectra mcp --agent NAME signs with the agent's key and forwards to the daemon (#9) eb675dfa5e18

//! `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 => {

            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::new(&ctx.socket(), 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) } }),

        ),

    }

}

//! The stdio shim, `spectra mcp --agent NAME` (spec §15.1), driven over

//! pipes as a client would: the handshake, tools, concurrency (a `wait` in

//! flight does not hold up other requests), nothing but protocol on stdout,

//! and plain answers when the daemon is not running.

mod common;

use common::{add_agent, hermetic, setup, Serving};

use serde_json::{json, Value};

use spectra_mcp::client::Client;

use std::process::Stdio;

use std::time::Duration;

use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, Lines};

use tokio::process::{ChildStdin, ChildStdout};

struct Shim {

    child: tokio::process::Child,

    stdin: ChildStdin,

    stdout: Lines<BufReader<ChildStdout>>,

}

impl Shim {

    fn start(d: &std::path::Path, agent: &str, extra: &[&str]) -> Shim {

        let mut cmd = tokio::process::Command::from(hermetic(d));

        cmd.args(["mcp", "--agent", agent])

            .args(extra)

            .stdin(Stdio::piped())

            .stdout(Stdio::piped())

            .stderr(Stdio::piped())

            .kill_on_drop(true);

        let mut child = cmd.spawn().unwrap();

        let stdin = child.stdin.take().unwrap();

        let stdout = BufReader::new(child.stdout.take().unwrap()).lines();

        Shim { child, stdin, stdout }

    }

    async fn send(&mut self, v: Value) {

        self.stdin.write_all(format!("{v}\n").as_bytes()).await.unwrap();

        self.stdin.flush().await.unwrap();

    }

    async fn recv(&mut self) -> Value {

        let line = tokio::time::timeout(Duration::from_secs(20), self.stdout.next_line())

            .await

            .expect("an answer within 20 s")

            .unwrap()

            .expect("a line");

        serde_json::from_str(&line).unwrap_or_else(|e| panic!("stdout carries only JSON-RPC, got {line:?}: {e}"))

    }

}

fn call(id: u64, name: &str, args: Value) -> Value {

    json!({ "jsonrpc": "2.0", "id": id, "method": "tools/call", "params": { "name": name, "arguments": args } })

}

#[tokio::test]

async fn a_client_speaks_to_spectra_through_the_shim() {

    let dir = setup();

    let d = dir.path();

    add_agent(d, "claude-oxygen", &[]);

    let kimi = add_agent(d, "kimi-oxygen", &[]);

    let srv = Serving::start(d);

    let mut s = Shim::start(d, "claude-oxygen", &[]);

    s.send(json!({ "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": { "name": "test", "version": "1" } } })).await;

    let init = s.recv().await;

    assert_eq!(init["result"]["protocolVersion"], "2025-06-18");

    s.send(json!({ "jsonrpc": "2.0", "method": "notifications/initialized" })).await;

    s.send(json!({ "jsonrpc": "2.0", "id": 2, "method": "tools/list" })).await;

    let list = s.recv().await;

    assert_eq!(list["id"], 2, "a notification gets no answer");

    assert_eq!(list["result"]["tools"].as_array().unwrap().len(), 11);

    s.send(call(3, "brief", json!({}))).await;

    let brief = s.recv().await;

    assert!(brief["result"]["content"][0]["text"].as_str().unwrap().contains("You are claude-oxygen · Andy McCall's agent"));

    // A wait in flight, then another request: the second is answered first.

    s.send(call(10, "wait", json!({ "timeout_s": 15 }))).await;

    tokio::time::sleep(Duration::from_millis(200)).await;

    s.send(json!({ "jsonrpc": "2.0", "id": 11, "method": "ping" })).await;

    assert_eq!(s.recv().await["id"], 11, "ping is not held up by the wait");

    let k = Client::new(&srv.socket, Some(kimi));

    let say = json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "say", "arguments": { "body": "@claude-oxygen ready when you are", "channel": "general" } } });

    assert_eq!(k.mcp("2025-11-25", &say).await.unwrap().status, 200);

    let woke = s.recv().await;

    assert_eq!(woke["id"], 10);

    assert!(woke["result"]["content"][0]["text"].as_str().unwrap().contains("ready when you are"), "{woke}");

    drop(s);

    srv.stop();

}

#[tokio::test]

async fn the_shim_says_plainly_what_is_wrong() {

    let dir = setup();

    let d = dir.path();

    add_agent(d, "claude-oxygen", &[]);

    add_agent(d, "kimi-oxygen", &[]);

    // No daemon: an answer, not silence.

    let mut s = Shim::start(d, "claude-oxygen", &[]);

    s.send(json!({ "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "2025-11-25" } })).await;

    let r = s.recv().await;

    assert!(r["error"]["message"].as_str().unwrap().contains("spectra serve"), "{r}");

    drop(s);

    let out = hermetic(d).args(["mcp", "--agent", "nobody"]).stdin(Stdio::null()).output().unwrap();

    assert!(!out.status.success() && String::from_utf8_lossy(&out.stderr).contains("no agent named nobody"));

    assert!(out.stdout.is_empty(), "errors never go to stdout");

    let other = d.join("agents/kimi-oxygen");

    let out =

        hermetic(d).args(["mcp", "--agent", "claude-oxygen", "--home", other.to_str().unwrap()]).stdin(Stdio::null()).output().unwrap();

    assert!(

        !out.status.success() && String::from_utf8_lossy(&out.stderr).contains("holds kimi-oxygen's key"),

        "{}",

        String::from_utf8_lossy(&out.stderr)

    );

}

- **Added: the stdio shim, `spectra mcp --agent NAME` (#9).** Any MCP client that starts stdio servers connects with one line and nothing secret in it: the shim signs every request with the agent's key from its own home and forwards it to the daemon. It speaks 2025-11-25, 2025-06-18 and 2025-03-26 (and passes 2026-07-28 through), answers requests concurrently so a `wait` holds nothing up, writes only protocol to stdout, and says plainly when the daemon is not running.

tokio = { workspace = true, features = ["io-std"] }

- [x] #9 The stdio shim — done 2026-11-02: spectra mcp --agent NAME, driven over pipes in tests/cli_mcp.rs. Real Claude Code, Antigravity and Kimi connections are checked in #13 with Andy, since they need his MCP configuration.

pub mod mcp;

    /// The stdio MCP server for one agent: signs with its key and forwards to the daemon

    Mcp {

        /// The agent it speaks for

        #[arg(long)]

        agent: String,

        /// The agent's home (default: the one recorded when it was added)

        #[arg(long)]

        home: Option<PathBuf>,

    },

        Command::Mcp { agent, home } => runtime()?.block_on(mcp::run(&ctx, &agent, home)),