| 1 |
|
| 2 |
|
| 3 |
|
| 4 |
|
| 5 |
|
| 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 |
|
| 19 |
|
| 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 |
|
| 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 |
|
| 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 |
} |