78d1f8a · sigitcodereview/sigit · siGit Code & Deploy

@@ -1,22 +1,20 @@

1 //! Model Context Protocol (MCP) client for siGit Code.

2 //!

3 -//! Implements the client half of the [Model Context Protocol](https://modelcontextprotocol.io):

4 -//! siGit Code connects to one or more MCP servers, discovers the tools they

5 -//! expose, and surfaces those tools to the model alongside its built-in ones.

6 -//! When the model calls an MCP tool, the call is forwarded to the owning server

7 -//! and the result fed back into the agent loop.

3 +//! siGit Code connects to one or more [MCP](https://modelcontextprotocol.io)

4 +//! servers, discovers the tools they expose, and surfaces those tools to the

5 +//! model alongside its built-in ones. When the model calls an MCP tool, the

6 +//! call is forwarded to the owning server and the result fed back into the

7 +//! agent loop.

8 //!

9 -//! Transports:

9 +//! The protocol itself is [`ed_mcp`]'s, which runs `rmcp`, the official Rust

10 +//! SDK. This module decides *which* servers to connect and how their tools are

11 +//! named and shown. Transports:

12 //!

11 -//! - **Streamable HTTP** — a single HTTP endpoint the client POSTs JSON-RPC 2.0

12 -//! messages to. The server answers either with a single `application/json`

13 -//! body or a `text/event-stream` (SSE) stream that carries the JSON-RPC

14 -//! response. Both are handled here. Configured with `url` in `mcp.toml`.

15 -//! - **stdio** — siGit spawns the server as a child process and exchanges

16 -//! newline-delimited JSON-RPC messages over its stdin/stdout (the server's

17 -//! stderr flows into siGit's own log stream). Configured with `command`

18 -//! (plus optional `args` and `[server.env]`) in `mcp.toml`. This is how most

19 -//! published MCP servers (filesystem, Playwright, GitHub, ...) are run.

13 +//! - **Streamable HTTP**, configured with `url` in `mcp.toml`.

14 +//! - **stdio**: siGit spawns the server as a child process (its stderr flows

15 +//! into siGit's own log stream). Configured with `command` (plus optional

16 +//! `args` and `[server.env]`) in `mcp.toml`. This is how most published MCP

17 +//! servers (filesystem, Playwright, GitHub, ...) are run.

18 //!

19 //! `url` and `command` are mutually exclusive; an entry with both, or neither,

20 //! is a config error that is logged and skipped.

@@ -47,9 +45,10 @@

45 //! stored in a process-global so the synchronous tool-spec builders

46 //! ([`tool_specs`]) and the async dispatch ([`call_tool`]) can both read it.

47 //!

50 -//! stdio children live for the sigit process. When a child dies (EOF or an I/O

51 -//! error on its pipes) the server is marked dead and later calls return an

52 -//! in-band error string the model can react to; there is no automatic restart.

48 +//! stdio children live for the sigit process. When a child dies, later calls

49 +//! return an in-band error string the model can react to; there is no

50 +//! automatic restart. An HTTP server that stops answering (a restart, an

51 +//! expired session) is reconnected once and the call retried.

52 //! `/reload` does *not* re-run discovery ([`init`] is once-per-process), so a

53 //! changed `mcp.toml` or a dead server needs a sigit restart. At process exit

54 //! children see EOF on their stdin and exit on their own.

@@ -63,18 +62,16 @@

62 //! are unused, so the dead-code lint is suppressed there only.

63 #![cfg_attr(not(unix), allow(dead_code))]

64

66 -use std::collections::{BTreeMap, HashMap};

65 +use std::collections::BTreeMap;

66 use std::path::PathBuf;

67 use std::sync::Mutex as StdMutex;

69 -use std::sync::atomic::{AtomicI64, Ordering};

68 use std::sync::{Arc, OnceLock};

69 use std::time::Duration;

70

71 +use ed_mcp::{ClientInfo, Connection, ServerSpec};

72 use serde::Deserialize;

73 use serde_json::{Value, json};

75 -use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};

76 -use tokio::process::{Child, ChildStdin, ChildStdout, Command};

77 -use tokio::sync::{Mutex, oneshot};

74 +use tokio::sync::RwLock;

75

76 use crate::backend::ToolSpec;

77

@@ -118,9 +115,6 @@ pub fn smbcloud_tool_suffix(name: &str) -> Option<&str> {

115 .strip_prefix("__")

116 }

117

121 -/// JSON-RPC / MCP protocol version we advertise in the handshake.

122 -const PROTOCOL_VERSION: &str = "2025-06-18";

123 -

118 /// Per-server budget for the connect + `initialize` + `tools/list` handshake at

119 /// startup. Bounds how long an unreachable server can delay startup; servers are

120 /// contacted concurrently, so this is the worst case for the whole set, not the

@@ -160,90 +154,19 @@ struct ServerConn {

154 /// Display endpoint for the `/mcp` listing: the URL for HTTP servers, the

155 /// command line for stdio servers.

156 endpoint: String,

163 - /// The live transport. `None` when a stdio server failed to even spawn.

164 - transport: Option<Transport>,

157 + /// How it was reached, kept so an HTTP server can be reconnected.

158 + def: ServerDef,

159 + /// The live connection. `None` when the server failed to connect.

160 + conn: RwLock<Option<Arc<Connection>>>,

161 /// Tools discovered at startup. Empty when the server failed to connect.

162 tools: Vec<McpTool>,

163 /// Connection error, if the handshake failed. Surfaced by `/mcp`.

164 error: Option<String>,

165 }

166

171 -/// How a connected server is reached.

172 -enum Transport {

173 - Http(HttpConn),

174 - Stdio(StdioConn),

175 -}

176 -

177 -/// Streamable HTTP connection state.

178 -struct HttpConn {

179 - /// Streamable HTTP endpoint (the single POST URL).

180 - url: String,

181 - /// Extra headers sent on every request (e.g. `Authorization`).

182 - headers: Vec<(String, String)>,

183 - /// Session id handed back by the server on `initialize`, echoed on every

184 - /// later request via the `Mcp-Session-Id` header.

185 - session_id: Mutex<Option<String>>,

186 -}

187 -

188 -/// stdio connection state: a child process speaking newline-delimited JSON-RPC

189 -/// over its stdin/stdout.

190 -struct StdioConn {

191 - /// The child's stdin. The mutex serializes writes so concurrent requests

192 - /// can't interleave bytes on the pipe; `None` once the pipe broke.

193 - writer: Mutex<Option<ChildStdin>>,

194 - /// State shared with the background reader task that owns the child's

195 - /// stdout.

196 - shared: Arc<StdioShared>,

197 - /// JSON-RPC id source. Ids are per-connection so the reader task can route

198 - /// each response to the request that carries its id.

199 - next_id: AtomicI64,

200 -}

201 -

202 -/// State shared between a [`StdioConn`] and its background reader task.

203 -struct StdioShared {

204 - /// Server name, for log lines.

205 - name: String,

206 - /// In-flight requests awaiting a response, keyed by JSON-RPC id. Dropping

207 - /// a sender (when the connection dies) wakes the waiter with an error.

208 - pending: StdMutex<HashMap<i64, oneshot::Sender<Value>>>,

209 - /// Why the connection is unusable, once it is (EOF, I/O error, kill).

210 - dead: StdMutex<Option<String>>,

211 - /// The child handle, kept so a dead/failed connection can kill and reap

212 - /// the process. Taken on death.

213 - child: StdMutex<Option<Child>>,

214 -}

215 -

216 -impl StdioShared {

217 - fn dead_reason(&self) -> Option<String> {

218 - self.dead.lock().unwrap().clone()

219 - }

220 -

221 - /// Mark the connection unusable: record the reason (first one wins), fail

222 - /// every in-flight request, and kill + reap the child, best effort.

223 - fn mark_dead(&self, reason: &str) {

224 - {

225 - let mut dead = self.dead.lock().unwrap();

226 - if dead.is_none() {

227 - *dead = Some(reason.to_string());

228 - }

229 - }

230 - // Dropping the senders wakes every waiter with a recv error.

231 - self.pending.lock().unwrap().clear();

232 - if let Some(mut child) = self.child.lock().unwrap().take() {

233 - let _ = child.start_kill();

234 - tokio::spawn(async move {

235 - let _ = child.wait().await;

236 - });

237 - }

238 - }

239 -}

240 -

241 -/// The process-global MCP state: a shared HTTP client plus every configured

242 -/// server.

167 +/// The process-global MCP state: every configured server.

168 struct Mcp {

244 - http: reqwest::Client,

169 servers: Vec<ServerConn>,

246 - next_id: AtomicI64,

170 }

171

172 static MCP: OnceLock<Mcp> = OnceLock::new();

@@ -377,6 +300,35 @@ struct ServerDef {

300 transport: TransportDef,

301 }

302

303 +impl ServerDef {

304 + /// The `ed-mcp` spec for this server. An `Authorization: Bearer` header is

305 + /// passed as the bearer token, which is how `rmcp` wants it.

306 + fn spec(&self) -> ServerSpec {

307 + match &self.transport {

308 + TransportDef::Http { url, headers } => {

309 + headers

310 + .iter()

311 + .fold(

312 + ServerSpec::http(&self.name, url),

313 + |spec, (key, value)| match value.strip_prefix("Bearer ") {

314 + Some(token) if key.eq_ignore_ascii_case("authorization") => {

315 + spec.with_bearer(token)

316 + }

317 + _ => spec.with_header(key, value),

318 + },

319 + )

320 + }

321 + TransportDef::Stdio { command, args, env } => {

322 + ServerSpec::stdio(&self.name, command, args.clone(), env.clone())

323 + }

324 + }

325 + }

326 +

327 + fn is_stdio(&self) -> bool {

328 + matches!(self.transport, TransportDef::Stdio { .. })

329 + }

330 +}

331 +

332 /// Config files to read, in priority order (later wins on a name clash):

333 /// global `$SIGIT_CONFIG_DIR/mcp.toml`, then project-local `<cwd>/.sigit/mcp.toml`.

334 fn config_paths() -> Vec<PathBuf> {

@@ -570,24 +522,9 @@ pub async fn init() {

522 return;

523 }

524

573 - let defs = load_configs();

574 - let http = reqwest::Client::builder()

575 - .timeout(CALL_TIMEOUT)

576 - .user_agent(concat!(

577 - "sigit/",

578 - env!("CARGO_PKG_VERSION"),

579 - " (mcp-client)"

580 - ))

581 - .build()

582 - .unwrap_or_default();

583 -

525 // Contact servers concurrently so one slow/unreachable host doesn't serialize

526 // the rest. Each handshake is bounded by HANDSHAKE_TIMEOUT.

586 - let connects = defs.into_iter().map(|def| {

587 - let http = http.clone();

588 - async move { connect(&http, def).await }

589 - });

590 - let servers = futures::future::join_all(connects).await;

527 + let servers = futures::future::join_all(load_configs().into_iter().map(connect)).await;

528

529 for server in &servers {

530 match &server.error {

@@ -600,154 +537,66 @@ pub async fn init() {

537 }

538 }

539

603 - let _ = MCP.set(Mcp {

604 - http,

605 - servers,

606 - next_id: AtomicI64::new(1),

607 - });

540 + let _ = MCP.set(Mcp { servers });

541 +}

542 +

543 +/// Sent to every server as `clientInfo`.

544 +fn client_info() -> ClientInfo {

545 + ClientInfo::new("sigit", env!("CARGO_PKG_VERSION"))

546 }

547

548 /// Run the handshake against one server and collect its tools. Always returns a

549 /// `ServerConn`; failures land in its `error` field rather than propagating.

612 -async fn connect(http: &reqwest::Client, def: ServerDef) -> ServerConn {

550 +async fn connect(def: ServerDef) -> ServerConn {

551 let endpoint = def.transport.endpoint();

614 - let transport = match &def.transport {

615 - TransportDef::Http { url, headers } => Transport::Http(HttpConn {

616 - url: url.clone(),

617 - headers: headers.clone(),

618 - session_id: Mutex::new(None),

619 - }),

620 - TransportDef::Stdio { command, args, env } => {

621 - match spawn_stdio(&def.name, command, args, env) {

622 - Ok(conn) => Transport::Stdio(conn),

623 - Err(error) => {

624 - return ServerConn {

625 - name: def.name,

626 - endpoint,

627 - transport: None,

628 - tools: Vec::new(),

629 - error: Some(error),

630 - };

631 - }

552 + let (conn, tools, error) =

553 + match Connection::connect(&def.spec(), &client_info(), HANDSHAKE_TIMEOUT).await {

554 + Ok(conn) => {

555 + let tools = server_tools(&def.name, &conn);

556 + (Some(Arc::new(conn)), tools, None)

557 }

633 - }

634 - };

635 -

636 - let mut conn = ServerConn {

637 - name: def.name,

558 + Err(error) => (None, Vec::new(), Some(format!("{error:#}"))),

559 + };

560 + ServerConn {

561 + name: def.name.clone(),

562 endpoint,

639 - transport: Some(transport),

640 - tools: Vec::new(),

641 - error: None,

642 - };

643 -

644 - let handshake = tokio::time::timeout(HANDSHAKE_TIMEOUT, async {

645 - // initialize → notifications/initialized → tools/list

646 - initialize(http, &conn).await?;

647 - notify_initialized(http, &conn).await?;

648 - list_tools(http, &conn).await

649 - })

650 - .await;

651 -

652 - match handshake {

653 - Ok(Ok(tools)) => conn.tools = tools,

654 - Ok(Err(error)) => conn.error = Some(error),

655 - Err(_) => conn.error = Some(format!("timed out after {}s", HANDSHAKE_TIMEOUT.as_secs())),

656 - }

657 -

658 - // A stdio child that failed its handshake is useless — kill it rather than

659 - // leave it running for the rest of the process.

660 - if let Some(error) = conn.error.clone()

661 - && let Some(Transport::Stdio(stdio)) = &conn.transport

662 - {

663 - stdio.shared.mark_dead(&error);

563 + def,

564 + conn: RwLock::new(conn),

565 + tools,

566 + error,

567 }

665 -

666 - conn

667 -}

668 -

669 -/// The `initialize` request: negotiate protocol version and (on HTTP) capture

670 -/// the session id from the response headers (handled inside [`post_rpc`]).

671 -async fn initialize(http: &reqwest::Client, conn: &ServerConn) -> Result<(), String> {

672 - let params = json!({

673 - "protocolVersion": PROTOCOL_VERSION,

674 - "capabilities": {},

675 - "clientInfo": { "name": "sigit", "version": env!("CARGO_PKG_VERSION") }

676 - });

677 - rpc_request(http, conn, "initialize", params, HANDSHAKE_TIMEOUT).await?;

678 - Ok(())

568 }

569

681 -/// The `notifications/initialized` notification. Servers expect it before

682 -/// fielding requests; it carries no id and no response.

683 -async fn notify_initialized(http: &reqwest::Client, conn: &ServerConn) -> Result<(), String> {

684 - rpc_notify(http, conn, "notifications/initialized", HANDSHAKE_TIMEOUT).await

685 -}

686 -

687 -/// `tools/list`, following `nextCursor` pagination, mapped into [`McpTool`]s.

688 -async fn list_tools(http: &reqwest::Client, conn: &ServerConn) -> Result<Vec<McpTool>, String> {

689 - let mut tools = Vec::new();

690 - let mut cursor: Option<String> = None;

691 -

692 - loop {

693 - let params = match &cursor {

694 - Some(c) => json!({ "cursor": c }),

695 - None => json!({}),

696 - };

697 - let result = rpc_request(http, conn, "tools/list", params, HANDSHAKE_TIMEOUT).await?;

698 -

699 - for tool in result

700 - .get("tools")

701 - .and_then(Value::as_array)

702 - .into_iter()

703 - .flatten()

704 - {

705 - let Some(remote_name) = tool.get("name").and_then(Value::as_str) else {

706 - continue;

707 - };

708 - let full_name = format!("{MCP_PREFIX}{}__{}", conn.name, sanitize(remote_name));

570 +/// The tools `conn` listed, in siGit's flattened form.

571 +fn server_tools(server: &str, conn: &Connection) -> Vec<McpTool> {

572 + conn.tools()

573 + .iter()

574 + .map(|tool| {

575 + let remote_name = tool.name.to_string();

576 + let full_name = format!("{MCP_PREFIX}{server}__{}", sanitize(&remote_name));

577 if full_name.chars().count() > 64 {

578 log::warn!(

579 "mcp: tool name '{full_name}' exceeds 64 chars; some backends may reject it"

580 );

581 }

714 - let remote_desc = tool

715 - .get("description")

716 - .and_then(Value::as_str)

717 - .unwrap_or("")

718 - .trim();

582 + let remote_desc = tool.description.as_deref().unwrap_or("").trim();

583 let description = if remote_desc.is_empty() {

720 - format!("[MCP server '{}'] {remote_name}", conn.name)

584 + format!("[MCP server '{server}'] {remote_name}")

585 } else {

722 - format!("[MCP server '{}'] {remote_desc}", conn.name)

586 + format!("[MCP server '{server}'] {remote_desc}")

587 };

724 - // `inputSchema` is a JSON Schema object; default to a permissive

725 - // object schema when a server omits it.

726 - let parameters_schema = tool

727 - .get("inputSchema")

728 - .filter(|schema| schema.is_object())

729 - .cloned()

730 - .unwrap_or_else(|| json!({ "type": "object" }))

731 - .to_string();

732 -

733 - tools.push(McpTool {

588 + let mut schema = Value::Object((*tool.input_schema).clone());

589 + if schema.get("type").is_none() {

590 + schema["type"] = json!("object");

591 + }

592 + McpTool {

593 full_name,

735 - remote_name: remote_name.to_string(),

594 + remote_name,

595 description,

737 - parameters_schema,

738 - });

739 - }

740 -

741 - cursor = result

742 - .get("nextCursor")

743 - .and_then(Value::as_str)

744 - .map(str::to_string);

745 - if cursor.is_none() {

746 - break;

747 - }

748 - }

749 -

750 - Ok(tools)

596 + parameters_schema: schema.to_string(),

597 + }

598 + })

599 + .collect()

600 }

601

602 // ── Client-supplied servers (per session) ───────────────────────────────────

@@ -780,8 +629,7 @@ pub enum ClientTransport {

629 /// Unlike the startup servers in [`MCP`], these belong to a session: the

630 /// client names them on `session/new` (or load/fork) and they are offered to

631 /// that session only. Cheap to clone; the connections are shared. When the

783 -/// last clone goes away the stdio children's stdin closes and they are killed;

784 -/// an HTTP server holds nothing open between calls.

632 +/// last clone goes away the connections close, which stops stdio children.

633 #[derive(Clone, Default)]

634 pub struct SessionServers(Arc<Vec<ServerConn>>);

635

@@ -817,7 +665,6 @@ pub async fn connect_session_servers(servers: Vec<ClientServer>) -> SessionServe

665 return SessionServers::default();

666 }

667

820 - let http = MCP.get().map(|mcp| mcp.http.clone()).unwrap_or_default();

668 let mut taken: Vec<String> = MCP

669 .get()

670 .map(|mcp| mcp.servers.iter().map(|s| s.name.clone()).collect())

@@ -836,13 +683,17 @@ pub async fn connect_session_servers(servers: Vec<ClientServer>) -> SessionServe

683 if name.is_empty() || taken.contains(&name) {

684 clashes.push(ServerConn {

685 endpoint: transport.endpoint(),

839 - transport: None,

840 - tools: Vec::new(),

686 error: Some(if name.is_empty() {

687 "the client sent a server with no name".to_string()

688 } else {

689 format!("another MCP server is already named '{name}'")

690 }),

691 + def: ServerDef {

692 + name: name.clone(),

693 + transport,

694 + },

695 + conn: RwLock::new(None),

696 + tools: Vec::new(),

697 name,

698 });

699 continue;

@@ -851,11 +702,7 @@ pub async fn connect_session_servers(servers: Vec<ClientServer>) -> SessionServe

702 defs.push(ServerDef { name, transport });

703 }

704

854 - let connects = defs.into_iter().map(|def| {

855 - let http = http.clone();

856 - async move { connect(&http, def).await }

857 - });

858 - let mut connected = futures::future::join_all(connects).await;

705 + let mut connected = futures::future::join_all(defs.into_iter().map(connect)).await;

706 connected.extend(clashes);

707

708 for server in &connected {

@@ -943,68 +790,59 @@ pub async fn call_tool(full_name: &str, arguments: &str) -> String {

790 }

791 };

792

946 - let http = MCP.get().map(|mcp| mcp.http.clone()).unwrap_or_default();

947 - match call_server(&http, server, &tool.remote_name, args).await {

793 + match call_server(server, &tool.remote_name, args).await {

794 Ok(text) => truncate(text),

795 Err(error) => format!("Error: {error}"),

796 }

797 }

798

953 -/// Send a `tools/call` and render the result into text. On HTTP, retries

954 -/// once after a re-`initialize` if the session was dropped (HTTP 404),

955 -/// which is how Streamable HTTP signals an expired session.

799 +/// Send a `tools/call` and render the result into text. An HTTP server whose

800 +/// call fails for any reason but a timeout is reconnected and the call retried

801 +/// once: that is how a restarted server or an expired session recovers.

802 async fn call_server(

957 - http: &reqwest::Client,

803 server: &ServerConn,

804 remote_name: &str,

805 args: Value,

806 ) -> Result<String, String> {

962 - let params = json!({ "name": remote_name, "arguments": args });

963 - let result = match &server.transport {

964 - None => return Err(format!("server '{}' is not connected", server.name)),

965 - Some(Transport::Stdio(stdio)) => {

966 - let timeout = if server.name == "xcode" {

967 - XCODE_CALL_TIMEOUT

968 - } else {

969 - CALL_TIMEOUT

970 - };

971 - stdio.request("tools/call", params, timeout).await?

807 + let timeout = if server.name == "xcode" {

808 + XCODE_CALL_TIMEOUT

809 + } else {

810 + CALL_TIMEOUT

811 + };

812 + let kind = if server.def.is_stdio() {

813 + "stdio server"

814 + } else {

815 + "server"

816 + };

817 + let Some(conn) = server.conn.read().await.clone() else {

818 + return Err(format!("{kind} '{}' is not connected", server.name));

819 + };

820 + let result = match conn.call_tool(remote_name, args.clone(), timeout).await {

821 + Ok(result) => result,

822 + Err(error) if server.def.is_stdio() || error.to_string().contains("timed out") => {

823 + return Err(format!("{kind} '{}': {error:#}", server.name));

824 }

973 - Some(Transport::Http(http_conn)) => {

974 - let body = json!({

975 - "jsonrpc": "2.0",

976 - "id": 0,

977 - "method": "tools/call",

978 - "params": params

979 - });

980 - let timeout = if server.name == "xcode" {

981 - XCODE_CALL_TIMEOUT

982 - } else {

983 - CALL_TIMEOUT

984 - };

985 - match post_rpc(http, &server.name, http_conn, &body, timeout).await {

986 - Ok(result) => result,

987 - Err(error) if error.contains("returned 404") => {

988 - // Session expired — drop it, re-handshake, and retry once.

989 - *http_conn.session_id.lock().await = None;

990 - initialize(http, server).await?;

991 - notify_initialized(http, server).await?;

992 - post_rpc(http, &server.name, http_conn, &body, timeout).await?

993 - }

994 - Err(error) => return Err(error),

995 - }

825 + Err(error) => {

826 + log::info!("mcp: '{}' failed ({error:#}); reconnecting", server.name);

827 + let fresh = Connection::connect(&server.def.spec(), &client_info(), HANDSHAKE_TIMEOUT)

828 + .await

829 + .map(Arc::new)

830 + .map_err(|again| {

831 + format!(

832 + "server '{}': {error:#}; reconnecting failed: {again:#}",

833 + server.name

834 + )

835 + })?;

836 + *server.conn.write().await = Some(fresh.clone());

837 + fresh

838 + .call_tool(remote_name, args, timeout)

839 + .await

840 + .map_err(|error| format!("server '{}': {error:#}", server.name))?

841 }

842 };

998 -

843 Ok(render_tool_result(&result))

844 }

845

1002 -impl Mcp {

1003 - fn next_id(&self) -> i64 {

1004 - self.next_id.fetch_add(1, Ordering::Relaxed)

1005 - }

1006 -}

1007 -

846 /// Flatten an MCP `tools/call` result into text. Joins text content blocks;

847 /// notes non-text blocks; honors `isError`.

848 fn render_tool_result(result: &Value) -> String {

@@ -1059,422 +897,6 @@ fn truncate(text: String) -> String {

897 format!("{kept}\n\n[output truncated to {RESULT_CHAR_LIMIT} characters]")

898 }

899

1062 -// ── Transport-generic JSON-RPC dispatch ─────────────────────────────────────

1063 -

1064 -/// Send a JSON-RPC request over whichever transport the server uses and return

1065 -/// its `result`.

1066 -async fn rpc_request(

1067 - http: &reqwest::Client,

1068 - conn: &ServerConn,

1069 - method: &str,

1070 - params: Value,

1071 - timeout: Duration,

1072 -) -> Result<Value, String> {

1073 - match &conn.transport {

1074 - None => Err(format!("server '{}' is not connected", conn.name)),

1075 - Some(Transport::Http(http_conn)) => {

1076 - let body = json!({ "jsonrpc": "2.0", "id": 0, "method": method, "params": params });

1077 - post_rpc(http, &conn.name, http_conn, &body, timeout).await

1078 - }

1079 - Some(Transport::Stdio(stdio)) => stdio.request(method, params, timeout).await,

1080 - }

1081 -}

1082 -

1083 -/// Send a JSON-RPC notification (no id, no response expected).

1084 -async fn rpc_notify(

1085 - http: &reqwest::Client,

1086 - conn: &ServerConn,

1087 - method: &str,

1088 - timeout: Duration,

1089 -) -> Result<(), String> {

1090 - match &conn.transport {

1091 - None => Err(format!("server '{}' is not connected", conn.name)),

1092 - Some(Transport::Http(http_conn)) => {

1093 - let body = json!({ "jsonrpc": "2.0", "method": method });

1094 - post_notification(http, &conn.name, http_conn, &body, timeout).await

1095 - }

1096 - Some(Transport::Stdio(stdio)) => stdio.notify(method).await,

1097 - }

1098 -}

1099 -

1100 -// ── stdio JSON-RPC plumbing ─────────────────────────────────────────────────

1101 -

1102 -/// Spawn a stdio MCP server and start its background reader task. The child's

1103 -/// stderr is inherited so it lands in sigit's own log stream; the given env

1104 -/// vars are added on top of the inherited environment.

1105 -fn spawn_stdio(

1106 - name: &str,

1107 - command: &str,

1108 - args: &[String],

1109 - env: &[(String, String)],

1110 -) -> Result<StdioConn, String> {

1111 - let mut cmd = Command::new(command);

1112 - cmd.args(args)

1113 - .stdin(std::process::Stdio::piped())

1114 - .stdout(std::process::Stdio::piped())

1115 - .stderr(std::process::Stdio::inherit())

1116 - .kill_on_drop(true);

1117 - for (key, value) in env {

1118 - cmd.env(key, value);

1119 - }

1120 - let mut child = cmd

1121 - .spawn()

1122 - .map_err(|error| format!("failed to spawn `{command}`: {error}"))?;

1123 - let stdin = child

1124 - .stdin

1125 - .take()

1126 - .ok_or_else(|| "child stdin was not captured".to_string())?;

1127 - let stdout = child

1128 - .stdout

1129 - .take()

1130 - .ok_or_else(|| "child stdout was not captured".to_string())?;

1131 -

1132 - let shared = Arc::new(StdioShared {

1133 - name: name.to_string(),

1134 - pending: StdMutex::new(HashMap::new()),

1135 - dead: StdMutex::new(None),

1136 - child: StdMutex::new(Some(child)),

1137 - });

1138 - tokio::spawn(stdio_reader(BufReader::new(stdout), Arc::clone(&shared)));

1139 -

1140 - Ok(StdioConn {

1141 - writer: Mutex::new(Some(stdin)),

1142 - shared,

1143 - next_id: AtomicI64::new(1),

1144 - })

1145 -}

1146 -

1147 -/// Background task owning a stdio child's stdout: parses one JSON-RPC message

1148 -/// per line and routes each response to the pending request that carries its

1149 -/// id. Server-initiated requests and notifications (anything with a `method`)

1150 -/// are logged and ignored — siGit doesn't support server→client calls. On EOF

1151 -/// or a read error the connection is marked dead, which fails every in-flight

1152 -/// request and reaps the child.

1153 -async fn stdio_reader(mut stdout: BufReader<ChildStdout>, shared: Arc<StdioShared>) {

1154 - let mut line = String::new();

1155 - loop {

1156 - line.clear();

1157 - match stdout.read_line(&mut line).await {

1158 - Ok(0) => {

1159 - shared.mark_dead("server closed its stdout (process exited)");

1160 - return;

1161 - }

1162 - Ok(_) => {}

1163 - Err(error) => {

1164 - shared.mark_dead(&format!("read error: {error}"));

1165 - return;

1166 - }

1167 - }

1168 - let trimmed = line.trim();

1169 - if trimmed.is_empty() {

1170 - continue;

1171 - }

1172 - let message: Value = match serde_json::from_str(trimmed) {

1173 - Ok(message) => message,

1174 - Err(error) => {

1175 - log::warn!("mcp: '{}' sent a non-JSON line: {error}", shared.name);

1176 - continue;

1177 - }

1178 - };

1179 - if let Some(method) = message.get("method").and_then(Value::as_str) {

1180 - log::debug!(

1181 - "mcp: ignoring server-initiated '{method}' from '{}'",

1182 - shared.name

1183 - );

1184 - continue;

1185 - }

1186 - let Some(id) = message

1187 - .get("id")

1188 - .and_then(|value| value.as_i64().or_else(|| value.as_str()?.parse().ok()))

1189 - else {

1190 - log::warn!(

1191 - "mcp: '{}' sent a response without a usable id; ignoring",

1192 - shared.name

1193 - );

1194 - continue;

1195 - };

1196 - let waiter = shared.pending.lock().unwrap().remove(&id);

1197 - match waiter {

1198 - Some(sender) => {

1199 - let _ = sender.send(message);

1200 - }

1201 - None => log::debug!(

1202 - "mcp: '{}' answered unknown/expired request id {id}; ignoring",

1203 - shared.name

1204 - ),

1205 - }

1206 - }

1207 -}

1208 -

1209 -impl Drop for StdioConn {

1210 - /// A connection nobody holds any more has no use for its child. Startup

1211 - /// servers live in a static and never get here; a session's servers do

1212 - /// once the session that brought them is replaced.

1213 - fn drop(&mut self) {

1214 - // `mark_dead` reaps on a spawned task, which needs a runtime.

1215 - if tokio::runtime::Handle::try_current().is_ok() {

1216 - self.shared.mark_dead("connection closed");

1217 - }

1218 - }

1219 -}

1220 -

1221 -impl StdioConn {

1222 - /// Send a JSON-RPC request and await its response, correlated by id. Fails

1223 - /// fast (in-band, never panicking) when the child has died.

1224 - async fn request(

1225 - &self,

1226 - method: &str,

1227 - params: Value,

1228 - timeout: Duration,

1229 - ) -> Result<Value, String> {

1230 - let name = &self.shared.name;

1231 - if let Some(reason) = self.shared.dead_reason() {

1232 - return Err(format!("stdio server '{name}' is not running: {reason}"));

1233 - }

1234 -

1235 - let id = self.next_id.fetch_add(1, Ordering::Relaxed);

1236 - let body = json!({ "jsonrpc": "2.0", "id": id, "method": method, "params": params });

1237 - let (sender, receiver) = oneshot::channel();

1238 - self.shared.pending.lock().unwrap().insert(id, sender);

1239 -

1240 - if let Err(error) = self.write_line(&body).await {

1241 - self.shared.pending.lock().unwrap().remove(&id);

1242 - self.shared.mark_dead(&error);

1243 - return Err(format!("stdio server '{name}': {error}"));

1244 - }

1245 -

1246 - let message = match tokio::time::timeout(timeout, receiver).await {

1247 - Ok(Ok(message)) => message,

1248 - // Our sender was dropped: the connection died mid-request.

1249 - Ok(Err(_)) => {

1250 - let reason = self

1251 - .shared

1252 - .dead_reason()

1253 - .unwrap_or_else(|| "connection closed".to_string());

1254 - return Err(format!("stdio server '{name}' is not running: {reason}"));

1255 - }

1256 - Err(_) => {

1257 - self.shared.pending.lock().unwrap().remove(&id);

1258 - return Err(format!(

1259 - "request to stdio server '{name}' timed out after {}s",

1260 - timeout.as_secs()

1261 - ));

1262 - }

1263 - };

1264 -

1265 - if let Some(error) = message.get("error") {

1266 - let code = error.get("code").and_then(Value::as_i64).unwrap_or(0);

1267 - let msg = error

1268 - .get("message")

1269 - .and_then(Value::as_str)

1270 - .unwrap_or("unknown error");

1271 - return Err(format!("'{name}' JSON-RPC error {code}: {msg}"));

1272 - }

1273 - message

1274 - .get("result")

1275 - .cloned()

1276 - .ok_or_else(|| format!("response from '{name}' had no result"))

1277 - }

1278 -

1279 - /// Send a JSON-RPC notification (no id, no response).

1280 - async fn notify(&self, method: &str) -> Result<(), String> {

1281 - let body = json!({ "jsonrpc": "2.0", "method": method });

1282 - if let Err(error) = self.write_line(&body).await {

1283 - self.shared.mark_dead(&error);

1284 - return Err(format!("stdio server '{}': {error}", self.shared.name));

1285 - }

1286 - Ok(())

1287 - }

1288 -

1289 - /// Write one newline-delimited JSON-RPC message. The writer mutex keeps

1290 - /// concurrent requests from interleaving bytes on the pipe.

1291 - async fn write_line(&self, body: &Value) -> Result<(), String> {

1292 - let mut guard = self.writer.lock().await;

1293 - let Some(writer) = guard.as_mut() else {

1294 - return Err("stdin already closed".to_string());

1295 - };

1296 - let mut line = body.to_string();

1297 - line.push('\n');

1298 - let result = async {

1299 - writer.write_all(line.as_bytes()).await?;

1300 - writer.flush().await

1301 - }

1302 - .await;

1303 - if let Err(error) = result {

1304 - // A broken pipe is unrecoverable; drop the writer so later calls

1305 - // fail fast.

1306 - *guard = None;

1307 - return Err(format!("write failed: {error}"));

1308 - }

1309 - Ok(())

1310 - }

1311 -}

1312 -

1313 -// ── Streamable HTTP JSON-RPC plumbing ───────────────────────────────────────

1314 -

1315 -/// POST a JSON-RPC request and return its `result`. Handles both an

1316 -/// `application/json` body and a `text/event-stream` (SSE) reply, captures the

1317 -/// session id from the response headers, and maps a JSON-RPC `error` to `Err`.

1318 -async fn post_rpc(

1319 - http: &reqwest::Client,

1320 - name: &str,

1321 - conn: &HttpConn,

1322 - body: &Value,

1323 - timeout: Duration,

1324 -) -> Result<Value, String> {

1325 - // Give every outbound request a fresh id; the on-the-wire id in `body` is a

1326 - // placeholder we overwrite so callers don't have to thread a counter.

1327 - let mut body = body.clone();

1328 - if body.get("id").is_some()

1329 - && let Some(mcp) = MCP.get()

1330 - {

1331 - body["id"] = json!(mcp.next_id());

1332 - }

1333 -

1334 - let response = build_request(http, conn, &body, timeout)

1335 - .await

1336 - .send()

1337 - .await

1338 - .map_err(|error| format!("request to {} failed: {error}", conn.url))?;

1339 -

1340 - // Persist the session id the server assigns on initialize.

1341 - if let Some(session) = response

1342 - .headers()

1343 - .get("mcp-session-id")

1344 - .and_then(|value| value.to_str().ok())

1345 - .map(str::to_string)

1346 - {

1347 - *conn.session_id.lock().await = Some(session);

1348 - }

1349 -

1350 - let status = response.status();

1351 - let content_type = response

1352 - .headers()

1353 - .get(reqwest::header::CONTENT_TYPE)

1354 - .and_then(|value| value.to_str().ok())

1355 - .unwrap_or("")

1356 - .to_string();

1357 -

1358 - if !status.is_success() {

1359 - let detail = response.text().await.unwrap_or_default();

1360 - let detail: String = detail.chars().take(500).collect();

1361 - return Err(format!(

1362 - "server '{name}' returned {}: {detail}",

1363 - status.as_u16()

1364 - ));

1365 - }

1366 -

1367 - let text = response

1368 - .text()

1369 - .await

1370 - .map_err(|error| format!("reading response from '{name}': {error}"))?;

1371 -

1372 - let message = if content_type.contains("text/event-stream") {

1373 - parse_sse_response(&text)

1374 - .ok_or_else(|| format!("no JSON-RPC message in SSE reply from '{name}'"))?

1375 - } else {

1376 - serde_json::from_str::<Value>(&text)

1377 - .map_err(|error| format!("parsing response from '{name}': {error}"))?

1378 - };

1379 -

1380 - if let Some(error) = message.get("error") {

1381 - let code = error.get("code").and_then(Value::as_i64).unwrap_or(0);

1382 - let msg = error

1383 - .get("message")

1384 - .and_then(Value::as_str)

1385 - .unwrap_or("unknown error");

1386 - return Err(format!("'{name}' JSON-RPC error {code}: {msg}"));

1387 - }

1388 -

1389 - message

1390 - .get("result")

1391 - .cloned()

1392 - .ok_or_else(|| format!("response from '{name}' had no result"))

1393 -}

1394 -

1395 -/// POST a JSON-RPC notification (no id, no response expected). A non-success

1396 -/// status is an error; an empty 202 body is the normal case.

1397 -async fn post_notification(

1398 - http: &reqwest::Client,

1399 - name: &str,

1400 - conn: &HttpConn,

1401 - body: &Value,

1402 - timeout: Duration,

1403 -) -> Result<(), String> {

1404 - let response = build_request(http, conn, body, timeout)

1405 - .await

1406 - .send()

1407 - .await

1408 - .map_err(|error| format!("notification to {} failed: {error}", conn.url))?;

1409 - if !response.status().is_success() {

1410 - return Err(format!(

1411 - "server '{name}' rejected notification: {}",

1412 - response.status().as_u16()

1413 - ));

1414 - }

1415 - Ok(())

1416 -}

1417 -

1418 -/// Build a request carrying the MCP headers: the dual `Accept`, the JSON body,

1419 -/// the configured static headers, the negotiated protocol version, and the

1420 -/// session id once we have one.

1421 -async fn build_request(

1422 - http: &reqwest::Client,

1423 - conn: &HttpConn,

1424 - body: &Value,

1425 - timeout: Duration,

1426 -) -> reqwest::RequestBuilder {

1427 - let mut request = http

1428 - .post(&conn.url)

1429 - .timeout(timeout)

1430 - .header(

1431 - reqwest::header::ACCEPT,

1432 - "application/json, text/event-stream",

1433 - )

1434 - .header("MCP-Protocol-Version", PROTOCOL_VERSION)

1435 - .json(body);

1436 -

1437 - for (key, value) in &conn.headers {

1438 - request = request.header(key.as_str(), value.as_str());

1439 - }

1440 - if let Some(session) = conn.session_id.lock().await.as_ref() {

1441 - request = request.header("Mcp-Session-Id", session.as_str());

1442 - }

1443 - request

1444 -}

1445 -

1446 -/// Extract the first JSON-RPC message from an SSE body. SSE frames are separated

1447 -/// by blank lines; each `data:` line contributes to the frame's payload. For a

1448 -/// single request/response exchange the server sends one `message` event whose

1449 -/// data is the JSON-RPC response.

1450 -fn parse_sse_response(body: &str) -> Option<Value> {

1451 - let mut data = String::new();

1452 - for line in body.lines() {

1453 - if let Some(rest) = line.strip_prefix("data:") {

1454 - if !data.is_empty() {

1455 - data.push('\n');

1456 - }

1457 - data.push_str(rest.strip_prefix(' ').unwrap_or(rest));

1458 - } else if line.trim().is_empty() && !data.is_empty() {

1459 - // End of an event — try to parse it as a JSON-RPC message.

1460 - if let Ok(value) = serde_json::from_str::<Value>(&data)

1461 - && (value.get("result").is_some() || value.get("error").is_some())

1462 - {

1463 - return Some(value);

1464 - }

1465 - data.clear();

1466 - }

1467 - }

1468 - // Trailing event without a closing blank line.

1469 - if !data.is_empty()

1470 - && let Ok(value) = serde_json::from_str::<Value>(&data)

1471 - && (value.get("result").is_some() || value.get("error").is_some())

1472 - {

1473 - return Some(value);

1474 - }

1475 - None

1476 -}

1477 -

900 // ── Status reporting (`/mcp`) ────────────────────────────────────────────────

901

902 /// Human-readable summary of configured MCP servers and their tools, for the

@@ -1747,27 +1169,6 @@ mod tests {

1169 assert_eq!(defs[0].transport.endpoint(), "u2");

1170 }

1171

1750 - #[test]

1751 - fn parse_sse_extracts_jsonrpc_response() {

1752 - let body =

1753 - "event: message\ndata: {\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"ok\":true}}\n\n";

1754 - let value = parse_sse_response(body).expect("a message");

1755 - assert_eq!(value["result"]["ok"], json!(true));

1756 - }

1757 -

1758 - #[test]

1759 - fn parse_sse_handles_no_trailing_blank_line() {

1760 - let body = "data: {\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{}}";

1761 - assert!(parse_sse_response(body).is_some());

1762 - }

1763 -

1764 - #[test]

1765 - fn parse_sse_ignores_non_response_frames() {

1766 - // A lone notification (no result/error) shouldn't be mistaken for the response.

1767 - let body = "data: {\"jsonrpc\":\"2.0\",\"method\":\"ping\"}\n\n";

1768 - assert!(parse_sse_response(body).is_none());

1769 - }

1770 -

1172 #[test]

1173 fn render_result_joins_text_blocks() {

1174 let result = json!({

Run MCP on ed-mcp instead of a hand-rolled JSON-RPC client