@sigitcodereview / sigit
Code Commits Issues Pulls Cloud Agent CI/CD
release/v1.6.1
claude/github-issue-40-0au5ua claude/linked-issue-acp-tui-u4vvpa claude/minor-release-cut-vkugow claude/sigit-acp-local-chat-cx6380 claude/sigit-cloud-agent-expansion-reox0a claude/tool-permission-system claude/zen-feynman-0u78dk development feature/acp-cancel-request feature/acp-session-list-cursor feature/acp-v1-conformance feature/agent-tools-multiedit-glob-todos-remember feature/cargo-workspace feature/hooks feature/load-local-model-explicitly feature/sigit-agent-profile feature/sigit-code-cloud feature/xcode-mcp-server fix/acp-replay-loaded-session-history fix/acp-replay-reasoning-and-attachments fix/dsml-doubled-bar-markers main release/v1.5.10 release/v1.5.6 release/v1.6.1
rs 1,815 lines 64.4 KB
Raw
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.
8 //!
9 //! Transports:
10 //!
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.
20 //!
21 //! `url` and `command` are mutually exclusive; an entry with both, or neither,
22 //! is a config error that is logged and skipped.
23 //!
24 //! ## Baked-in servers
25 //!
26 //! siGit Code bakes in its official MCP server at `<cloud>/mcp` (default
27 //! `https://sigit.si/api/v1/mcp`, following `SIGIT_CLOUD_URL`). When the user is
28 //! signed in (`sigit login`) the cloud session token is sent as a bearer
29 //! credential.
30 //!
31 //! The smbCloud CLI's MCP server (`smb --mcp`, stdio) is also baked in, but
32 //! only when the `smb` binary is actually on `PATH` — no binary, no entry, no
33 //! error. Opt out with `smbcloud = false` in `mcp.toml` or
34 //! `SIGIT_MCP_SMBCLOUD=off`.
35 //!
36 //! Both baked-in entries yield to a user-defined `mcp.toml` server of the same
37 //! name, so either can be repointed or reconfigured without a special case.
38 //! Additional servers are configured in `mcp.toml` (see [`load_configs`]).
39 //!
40 //! ## Lifecycle
41 //!
42 //! Discovery is best-effort and happens once at startup via [`init`]: each
43 //! configured server is contacted concurrently (with a per-server timeout),
44 //! runs the `initialize` handshake, and has its `tools/list` cached. A server
45 //! that fails to connect is recorded with its error and simply contributes no
46 //! tools — it never blocks startup or the rest of the agent. The result is
47 //! stored in a process-global so the synchronous tool-spec builders
48 //! ([`tool_specs`]) and the async dispatch ([`call_tool`]) can both read it.
49 //!
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.
53 //! `/reload` does *not* re-run discovery ([`init`] is once-per-process), so a
54 //! changed `mcp.toml` or a dead server needs a sigit restart. At process exit
55 //! children see EOF on their stdin and exit on their own.
56 //!
57 //! Tools are namespaced `mcp__<server>__<tool>` so they never collide with
58 //! built-in tools or with each other across servers. This mirrors the
59 //! convention used by other MCP-aware agents.
60 //!
61 //! Like the rest of the backend seam, MCP is wired up only through the
62 //! interactive client and the ACP agent loop. On non-Unix targets a few helpers
63 //! are unused, so the dead-code lint is suppressed there only.
64 #![cfg_attr(not(unix), allow(dead_code))]
65
66 use std::collections::{BTreeMap, HashMap};
67 use std::path::PathBuf;
68 use std::sync::Mutex as StdMutex;
69 use std::sync::atomic::{AtomicI64, Ordering};
70 use std::sync::{Arc, OnceLock};
71 use std::time::Duration;
72
73 use serde::Deserialize;
74 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};
78
79 use crate::backend::ToolSpec;
80
81 /// Prefix marking a tool as MCP-provided. The full name is
82 /// `mcp__<server>__<tool>`.
83 pub const MCP_PREFIX: &str = "mcp__";
84
85 /// Name of the baked-in official siGit Code server; its tools are namespaced
86 /// `mcp__sigit__<tool>`. A user-defined `mcp.toml` entry with this name
87 /// overrides the baked-in URL/headers but keeps the namespace, so callers of
88 /// [`official_tool_name`] reach whatever the user pointed `sigit` at.
89 pub const OFFICIAL_SERVER_NAME: &str = "sigit";
90
91 /// The full namespaced name of a tool on the official server, e.g.
92 /// `official_tool_name("list_issues")` → `mcp__sigit__list_issues`.
93 pub fn official_tool_name(tool: &str) -> String {
94 format!("{MCP_PREFIX}{OFFICIAL_SERVER_NAME}__{tool}")
95 }
96
97 /// The bare tool name when `name` belongs to the official server
98 /// (`mcp__sigit__list_issues` → `Some("list_issues")`), else `None`.
99 pub fn official_tool_suffix(name: &str) -> Option<&str> {
100 name.strip_prefix(MCP_PREFIX)?
101 .strip_prefix(OFFICIAL_SERVER_NAME)?
102 .strip_prefix("__")
103 }
104
105 /// Name of the baked-in smbCloud CLI server; its tools are namespaced
106 /// `mcp__smbcloud__<tool>`. Like the official server, a user-defined `mcp.toml`
107 /// entry with this name overrides the baked-in command line.
108 const SMBCLOUD_SERVER_NAME: &str = "smbcloud";
109
110 /// The smbCloud CLI binary the baked-in stdio entry spawns (`smb --mcp`).
111 const SMBCLOUD_COMMAND: &str = "smb";
112
113 /// The bare tool name when `name` belongs to the smbCloud server
114 /// (`mcp__smbcloud__project_list` → `Some("project_list")`), else `None`.
115 pub fn smbcloud_tool_suffix(name: &str) -> Option<&str> {
116 name.strip_prefix(MCP_PREFIX)?
117 .strip_prefix(SMBCLOUD_SERVER_NAME)?
118 .strip_prefix("__")
119 }
120
121 /// JSON-RPC / MCP protocol version we advertise in the handshake.
122 const PROTOCOL_VERSION: &str = "2025-06-18";
123
124 /// Per-server budget for the connect + `initialize` + `tools/list` handshake at
125 /// startup. Bounds how long an unreachable server can delay startup; servers are
126 /// contacted concurrently, so this is the worst case for the whole set, not the
127 /// sum.
128 const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(8);
129
130 /// Overall request timeout for an individual `tools/call`. Generous for build
131 /// and test tools, while the Xcode bridge gets a shorter bound below because
132 /// it otherwise leaves an ACP prompt looking permanently busy when Xcode
133 /// cannot service the request.
134 const CALL_TIMEOUT: Duration = Duration::from_secs(120);
135 const XCODE_CALL_TIMEOUT: Duration = Duration::from_secs(30);
136
137 /// Cap on the characters returned from a single tool call, so a chatty server
138 /// can't blow up the model's context. Matches the spirit of the file-read cap.
139 const RESULT_CHAR_LIMIT: usize = 30_000;
140
141 // ── Public types ──────────────────────────────────────────────────────────────
142
143 /// A tool discovered on an MCP server, in siGit's flattened form.
144 #[derive(Debug, Clone)]
145 struct McpTool {
146 /// Namespaced name exposed to the model: `mcp__<server>__<tool>`.
147 full_name: String,
148 /// The tool's name as the server knows it (sent back in `tools/call`).
149 remote_name: String,
150 /// Human/model-facing description, prefixed with the server name.
151 description: String,
152 /// JSON Schema for the tool's arguments, encoded as a string.
153 parameters_schema: String,
154 }
155
156 /// A configured MCP server and its live connection state.
157 struct ServerConn {
158 /// Sanitized server name used in tool namespacing and the `/mcp` listing.
159 name: String,
160 /// Display endpoint for the `/mcp` listing: the URL for HTTP servers, the
161 /// command line for stdio servers.
162 endpoint: String,
163 /// The live transport. `None` when a stdio server failed to even spawn.
164 transport: Option<Transport>,
165 /// Tools discovered at startup. Empty when the server failed to connect.
166 tools: Vec<McpTool>,
167 /// Connection error, if the handshake failed. Surfaced by `/mcp`.
168 error: Option<String>,
169 }
170
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.
243 struct Mcp {
244 http: reqwest::Client,
245 servers: Vec<ServerConn>,
246 next_id: AtomicI64,
247 }
248
249 static MCP: OnceLock<Mcp> = OnceLock::new();
250
251 // ── Configuration ───────────────────────────────────────────────────────────
252
253 /// Default endpoint of the official siGit Code MCP server, derived from the
254 /// cloud base URL so `SIGIT_CLOUD_URL` (dev) carries over.
255 fn official_url() -> String {
256 format!(
257 "{}/mcp",
258 crate::provider::cloud_base_url().trim_end_matches('/')
259 )
260 }
261
262 /// A server entry as written in `mcp.toml`. Exactly one of `url` (Streamable
263 /// HTTP) or `command` (stdio) selects the transport.
264 #[derive(Debug, Deserialize)]
265 struct ServerEntry {
266 name: String,
267 /// Streamable HTTP endpoint. Mutually exclusive with `command`.
268 #[serde(default)]
269 url: Option<String>,
270 /// stdio server executable. Mutually exclusive with `url`.
271 #[serde(default)]
272 command: Option<String>,
273 /// Arguments for `command`.
274 #[serde(default)]
275 args: Vec<String>,
276 /// Extra environment variables for `command`, added on top of the
277 /// inherited environment.
278 #[serde(default)]
279 env: BTreeMap<String, String>,
280 /// Set `enabled = false` to keep an entry in the file but skip connecting.
281 #[serde(default)]
282 enabled: Option<bool>,
283 /// Static headers, e.g. `Authorization = "Bearer ..."`. HTTP only.
284 #[serde(default)]
285 headers: BTreeMap<String, String>,
286 }
287
288 impl ServerEntry {
289 /// Resolve the entry's transport. `url` and `command` are mutually
290 /// exclusive and exactly one is required; anything else is a config error.
291 fn transport_def(&self) -> Result<TransportDef, String> {
292 let url = self.url.as_deref().map(str::trim).filter(|v| !v.is_empty());
293 let command = self
294 .command
295 .as_deref()
296 .map(str::trim)
297 .filter(|v| !v.is_empty());
298 match (url, command) {
299 (Some(_), Some(_)) => {
300 Err("has both `url` and `command`; a server uses exactly one transport".to_string())
301 }
302 (None, None) => {
303 Err("needs either `url` (Streamable HTTP) or `command` (stdio)".to_string())
304 }
305 (Some(url), None) => Ok(TransportDef::Http {
306 url: url.to_string(),
307 headers: self
308 .headers
309 .iter()
310 .map(|(k, v)| (k.clone(), v.clone()))
311 .collect(),
312 }),
313 (None, Some(command)) => Ok(TransportDef::Stdio {
314 command: command.to_string(),
315 args: self.args.clone(),
316 env: self
317 .env
318 .iter()
319 .map(|(k, v)| (k.clone(), v.clone()))
320 .collect(),
321 }),
322 }
323 }
324 }
325
326 /// The `mcp.toml` schema.
327 #[derive(Debug, Default, Deserialize)]
328 struct McpFile {
329 /// Include the baked-in official server. Defaults to `true`; set `false` to
330 /// opt out.
331 #[serde(default)]
332 official: Option<bool>,
333 /// Include the baked-in smbCloud CLI server (`smb --mcp`). Defaults to
334 /// `true`; set `false` to opt out. Moot when `smb` isn't installed.
335 #[serde(default)]
336 smbcloud: Option<bool>,
337 #[serde(default)]
338 server: Vec<ServerEntry>,
339 }
340
341 /// How to reach a configured server, before connecting.
342 #[derive(Debug, Clone)]
343 enum TransportDef {
344 Http {
345 url: String,
346 headers: Vec<(String, String)>,
347 },
348 Stdio {
349 command: String,
350 args: Vec<String>,
351 env: Vec<(String, String)>,
352 },
353 }
354
355 impl TransportDef {
356 /// Human-readable endpoint for logs and the `/mcp` listing: the URL for
357 /// HTTP, the command line for stdio.
358 fn endpoint(&self) -> String {
359 match self {
360 TransportDef::Http { url, .. } => url.clone(),
361 TransportDef::Stdio { command, args, .. } => {
362 let mut line = command.clone();
363 for arg in args {
364 line.push(' ');
365 line.push_str(arg);
366 }
367 line
368 }
369 }
370 }
371 }
372
373 /// A resolved server definition, before connecting.
374 #[derive(Debug, Clone)]
375 struct ServerDef {
376 name: String,
377 transport: TransportDef,
378 }
379
380 /// Config files to read, in priority order (later wins on a name clash):
381 /// global `$SIGIT_CONFIG_DIR/mcp.toml`, then project-local `<cwd>/.sigit/mcp.toml`.
382 fn config_paths() -> Vec<PathBuf> {
383 let mut paths = Vec::new();
384 if let Some(dir) = sigit_config_dir() {
385 paths.push(dir.join("mcp.toml"));
386 }
387 if let Ok(cwd) = std::env::current_dir() {
388 paths.push(cwd.join(".sigit").join("mcp.toml"));
389 }
390 paths
391 }
392
393 fn sigit_config_dir() -> Option<PathBuf> {
394 if let Ok(dir) = std::env::var("SIGIT_CONFIG_DIR")
395 && !dir.is_empty()
396 {
397 return Some(PathBuf::from(dir));
398 }
399 std::env::var("HOME")
400 .ok()
401 .map(|home| PathBuf::from(home).join(".config").join("sigit"))
402 }
403
404 /// Global escape hatch: `SIGIT_MCP=off` disables MCP entirely, including
405 /// servers an ACP client supplies per session.
406 fn disabled_by_env() -> bool {
407 std::env::var("SIGIT_MCP").is_ok_and(|value| {
408 matches!(
409 value.trim().to_ascii_lowercase().as_str(),
410 "off" | "0" | "false" | "no" | "disabled"
411 )
412 })
413 }
414
415 /// Resolve the full set of servers to connect to: the baked-in official server
416 /// (unless opted out) plus any from `mcp.toml`. Project-local entries override
417 /// global ones, and a user entry named `sigit` overrides the official default.
418 fn load_configs() -> Vec<ServerDef> {
419 if disabled_by_env() {
420 log::info!("mcp: disabled via SIGIT_MCP");
421 return Vec::new();
422 }
423
424 let mut include_official = true;
425 let mut include_smbcloud = true;
426 // De-duplicated by sanitized name; a later config file overrides an earlier
427 // one for the same name (project-local wins over global).
428 let mut defs: Vec<ServerDef> = Vec::new();
429
430 for path in config_paths() {
431 let Ok(contents) = std::fs::read_to_string(&path) else {
432 continue;
433 };
434 let parsed: McpFile = match toml::from_str(&contents) {
435 Ok(parsed) => parsed,
436 Err(error) => {
437 log::warn!("mcp: ignoring {}: {error}", path.display());
438 continue;
439 }
440 };
441 if let Some(official) = parsed.official {
442 include_official = official;
443 }
444 if let Some(smbcloud) = parsed.smbcloud {
445 include_smbcloud = smbcloud;
446 }
447 for entry in parsed.server {
448 if entry.enabled == Some(false) {
449 continue;
450 }
451 let name = sanitize(&entry.name);
452 if name.is_empty() {
453 log::warn!("mcp: skipping server with empty name in {}", path.display());
454 continue;
455 }
456 let transport = match entry.transport_def() {
457 Ok(transport) => transport,
458 Err(error) => {
459 log::warn!(
460 "mcp: skipping server '{name}' in {}: {error}",
461 path.display()
462 );
463 continue;
464 }
465 };
466 upsert(&mut defs, ServerDef { name, transport });
467 }
468 }
469
470 // The official server can also be disabled with SIGIT_MCP_OFFICIAL=off.
471 if let Ok(value) = std::env::var("SIGIT_MCP_OFFICIAL")
472 && matches!(
473 value.trim().to_ascii_lowercase().as_str(),
474 "off" | "0" | "false" | "no"
475 )
476 {
477 include_official = false;
478 }
479
480 // Add the baked-in official server, but never clobber a user-defined entry
481 // named `sigit` — an explicit config (e.g. a custom URL or headers) wins.
482 if include_official && !defs.iter().any(|d| d.name == OFFICIAL_SERVER_NAME) {
483 let mut headers = Vec::new();
484 if let Some(token) = crate::credentials::load_token() {
485 headers.push(("Authorization".to_string(), format!("Bearer {token}")));
486 }
487 defs.push(ServerDef {
488 name: OFFICIAL_SERVER_NAME.to_string(),
489 transport: TransportDef::Http {
490 url: official_url(),
491 headers,
492 },
493 });
494 }
495
496 // The smbCloud server can also be disabled with SIGIT_MCP_SMBCLOUD=off.
497 if let Ok(value) = std::env::var("SIGIT_MCP_SMBCLOUD")
498 && matches!(
499 value.trim().to_ascii_lowercase().as_str(),
500 "off" | "0" | "false" | "no"
501 )
502 {
503 include_smbcloud = false;
504 }
505
506 // Add the baked-in smbCloud CLI server, with the same never-clobber rule as
507 // the official one. Only when `smb` is actually installed: a hardwired
508 // entry for a binary most users don't have would surface a spawn failure
509 // in `/mcp` instead of just staying out of the way.
510 if include_smbcloud && !defs.iter().any(|d| d.name == SMBCLOUD_SERVER_NAME) {
511 if on_path(SMBCLOUD_COMMAND) {
512 defs.push(ServerDef {
513 name: SMBCLOUD_SERVER_NAME.to_string(),
514 transport: TransportDef::Stdio {
515 command: SMBCLOUD_COMMAND.to_string(),
516 args: vec!["--mcp".to_string()],
517 env: Vec::new(),
518 },
519 });
520 } else {
521 log::debug!("mcp: `{SMBCLOUD_COMMAND}` not on PATH; skipping the smbcloud server");
522 }
523 }
524
525 defs
526 }
527
528 /// Whether `binary` resolves to an executable file on `PATH` (with the
529 /// platform's executable suffix, `.exe` on Windows).
530 fn on_path(binary: &str) -> bool {
531 let Some(path) = std::env::var_os("PATH") else {
532 return false;
533 };
534 let file = format!("{binary}{}", std::env::consts::EXE_SUFFIX);
535 std::env::split_paths(&path).any(|dir| !dir.as_os_str().is_empty() && dir.join(&file).is_file())
536 }
537
538 /// Insert `def`, replacing any existing entry with the same name.
539 fn upsert(defs: &mut Vec<ServerDef>, def: ServerDef) {
540 if let Some(slot) = defs.iter_mut().find(|d| d.name == def.name) {
541 *slot = def;
542 } else {
543 defs.push(def);
544 }
545 }
546
547 /// Sanitize a name into the `[a-zA-Z0-9_-]` set tool names are restricted to,
548 /// collapsing anything else to `_`.
549 fn sanitize(raw: &str) -> String {
550 raw.trim()
551 .chars()
552 .map(|c| {
553 if c.is_ascii_alphanumeric() || c == '_' || c == '-' {
554 c
555 } else {
556 '_'
557 }
558 })
559 .collect()
560 }
561
562 // ── Startup / discovery ─────────────────────────────────────────────────────
563
564 /// Connect to every configured server and cache the tools they expose. Idempotent
565 /// and best-effort: a server that can't be reached is recorded with its error and
566 /// contributes no tools. Safe to call from either entry point; only the first
567 /// call does work.
568 pub async fn init() {
569 if MCP.get().is_some() {
570 return;
571 }
572
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
584 // Contact servers concurrently so one slow/unreachable host doesn't serialize
585 // 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;
591
592 for server in &servers {
593 match &server.error {
594 Some(error) => log::warn!("mcp: server '{}' unavailable: {error}", server.name),
595 None => log::info!(
596 "mcp: server '{}' ready, {} tool(s)",
597 server.name,
598 server.tools.len()
599 ),
600 }
601 }
602
603 let _ = MCP.set(Mcp {
604 http,
605 servers,
606 next_id: AtomicI64::new(1),
607 });
608 }
609
610 /// Run the handshake against one server and collect its tools. Always returns a
611 /// `ServerConn`; failures land in its `error` field rather than propagating.
612 async fn connect(http: &reqwest::Client, def: ServerDef) -> ServerConn {
613 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 }
632 }
633 }
634 };
635
636 let mut conn = ServerConn {
637 name: def.name,
638 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);
664 }
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(())
679 }
680
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));
709 if full_name.chars().count() > 64 {
710 log::warn!(
711 "mcp: tool name '{full_name}' exceeds 64 chars; some backends may reject it"
712 );
713 }
714 let remote_desc = tool
715 .get("description")
716 .and_then(Value::as_str)
717 .unwrap_or("")
718 .trim();
719 let description = if remote_desc.is_empty() {
720 format!("[MCP server '{}'] {remote_name}", conn.name)
721 } else {
722 format!("[MCP server '{}'] {remote_desc}", conn.name)
723 };
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 {
734 full_name,
735 remote_name: remote_name.to_string(),
736 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)
751 }
752
753 // ── Client-supplied servers (per session) ───────────────────────────────────
754
755 /// A server an ACP client passed in `mcpServers` on a session request.
756 #[derive(Debug, Clone)]
757 pub struct ClientServer {
758 pub name: String,
759 pub transport: ClientTransport,
760 }
761
762 /// How a client-supplied server is reached. stdio is the transport every ACP
763 /// agent must support; Streamable HTTP is the optional one siGit Code
764 /// advertises through `mcpCapabilities.http`.
765 #[derive(Debug, Clone)]
766 pub enum ClientTransport {
767 Stdio {
768 command: String,
769 args: Vec<String>,
770 env: Vec<(String, String)>,
771 },
772 Http {
773 url: String,
774 headers: Vec<(String, String)>,
775 },
776 }
777
778 /// The MCP servers one ACP session brought with it, connected.
779 ///
780 /// Unlike the startup servers in [`MCP`], these belong to a session: the
781 /// client names them on `session/new` (or load/fork) and they are offered to
782 /// 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.
785 #[derive(Clone, Default)]
786 pub struct SessionServers(Arc<Vec<ServerConn>>);
787
788 impl std::fmt::Debug for SessionServers {
789 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
790 f.debug_list()
791 .entries(self.0.iter().map(|server| &server.name))
792 .finish()
793 }
794 }
795
796 /// The live session's client-supplied servers. One process serves every
797 /// thread the editor has open, so like the roots in `workspace.rs` this
798 /// always belongs to the session `main.rs` has installed.
799 static SESSION_SERVERS: StdMutex<Option<SessionServers>> = StdMutex::new(None);
800
801 /// Connect to the servers a client supplied for one session. Best effort, like
802 /// [`init`]: a server that fails its handshake is kept with its error so `/mcp`
803 /// can show it, and contributes no tools.
804 ///
805 /// A name already taken by a startup server (or by an earlier entry in the same
806 /// list) is not connected. Its tools would share the `mcp__<server>__` prefix
807 /// with the other server's, and a call could not be routed.
808 pub async fn connect_session_servers(servers: Vec<ClientServer>) -> SessionServers {
809 if servers.is_empty() {
810 return SessionServers::default();
811 }
812 if disabled_by_env() {
813 log::info!(
814 "mcp: ignoring {} client-supplied server(s), disabled via SIGIT_MCP",
815 servers.len()
816 );
817 return SessionServers::default();
818 }
819
820 let http = MCP.get().map(|mcp| mcp.http.clone()).unwrap_or_default();
821 let mut taken: Vec<String> = MCP
822 .get()
823 .map(|mcp| mcp.servers.iter().map(|s| s.name.clone()).collect())
824 .unwrap_or_default();
825
826 let mut clashes = Vec::new();
827 let mut defs = Vec::new();
828 for server in servers {
829 let name = sanitize(&server.name);
830 let transport = match server.transport {
831 ClientTransport::Stdio { command, args, env } => {
832 TransportDef::Stdio { command, args, env }
833 }
834 ClientTransport::Http { url, headers } => TransportDef::Http { url, headers },
835 };
836 if name.is_empty() || taken.contains(&name) {
837 clashes.push(ServerConn {
838 endpoint: transport.endpoint(),
839 transport: None,
840 tools: Vec::new(),
841 error: Some(if name.is_empty() {
842 "the client sent a server with no name".to_string()
843 } else {
844 format!("another MCP server is already named '{name}'")
845 }),
846 name,
847 });
848 continue;
849 }
850 taken.push(name.clone());
851 defs.push(ServerDef { name, transport });
852 }
853
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;
859 connected.extend(clashes);
860
861 for server in &connected {
862 match &server.error {
863 Some(error) => log::warn!(
864 "mcp: client-supplied server '{}' unavailable: {error}",
865 server.name
866 ),
867 None => log::info!(
868 "mcp: client-supplied server '{}' ready, {} tool(s)",
869 server.name,
870 server.tools.len()
871 ),
872 }
873 }
874
875 SessionServers(Arc::new(connected))
876 }
877
878 /// Make `servers` the live session's set. Called whenever `main.rs` installs a
879 /// session, so a thread never sees another thread's servers.
880 pub fn set_session_servers(servers: SessionServers) {
881 *SESSION_SERVERS.lock().unwrap() = Some(servers);
882 }
883
884 fn session_servers() -> SessionServers {
885 SESSION_SERVERS.lock().unwrap().clone().unwrap_or_default()
886 }
887
888 // ── Tool exposure + dispatch ────────────────────────────────────────────────
889
890 /// Whether a tool name belongs to MCP. The dispatch in `tools::execute_tool`
891 /// uses this to route a call here.
892 pub fn is_mcp_tool(name: &str) -> bool {
893 name.starts_with(MCP_PREFIX)
894 }
895
896 /// All discovered MCP tools as agent [`ToolSpec`]s, ready to append to the
897 /// built-in tool list: the startup servers' tools, then the ones the live
898 /// session's client supplied. Empty when no server exposed any.
899 pub fn tool_specs() -> Vec<ToolSpec> {
900 let session = session_servers();
901 let startup = MCP.get().map(|mcp| mcp.servers.as_slice()).unwrap_or(&[]);
902 let mut specs = Vec::new();
903 for server in startup.iter().chain(session.0.iter()) {
904 for tool in &server.tools {
905 specs.push(ToolSpec {
906 name: tool.full_name.clone(),
907 description: tool.description.clone(),
908 parameters_schema: tool.parameters_schema.clone(),
909 });
910 }
911 }
912 specs
913 }
914
915 /// Execute an MCP tool call by name, returning text to feed back to the model.
916 /// Errors are returned as plain strings (never panics) so a failing tool degrades
917 /// to a message the model can react to, exactly like the built-in tools.
918 pub async fn call_tool(full_name: &str, arguments: &str) -> String {
919 let session = session_servers();
920 let startup = MCP.get().map(|mcp| mcp.servers.as_slice()).unwrap_or(&[]);
921 if MCP.get().is_none() && session.0.is_empty() {
922 return "Error: MCP is not initialized.".to_string();
923 }
924
925 let Some((server, tool)) = startup.iter().chain(session.0.iter()).find_map(|s| {
926 s.tools
927 .iter()
928 .find(|t| t.full_name == full_name)
929 .map(|t| (s, t))
930 }) else {
931 return format!("Error: unknown MCP tool \"{full_name}\".");
932 };
933
934 // Arguments arrive as a JSON-encoded string; an empty/blank string means no
935 // arguments. Anything that isn't a JSON object is a model mistake.
936 let args: Value = if arguments.trim().is_empty() {
937 json!({})
938 } else {
939 match serde_json::from_str(arguments) {
940 Ok(value @ Value::Object(_)) => value,
941 Ok(_) => return "Error: tool arguments must be a JSON object.".to_string(),
942 Err(error) => return format!("Error: failed to parse arguments: {error}"),
943 }
944 };
945
946 let http = MCP.get().map(|mcp| mcp.http.clone()).unwrap_or_default();
947 match call_server(&http, server, &tool.remote_name, args).await {
948 Ok(text) => truncate(text),
949 Err(error) => format!("Error: {error}"),
950 }
951 }
952
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.
956 async fn call_server(
957 http: &reqwest::Client,
958 server: &ServerConn,
959 remote_name: &str,
960 args: Value,
961 ) -> 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?
972 }
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 }
996 }
997 };
998
999 Ok(render_tool_result(&result))
1000 }
1001
1002 impl Mcp {
1003 fn next_id(&self) -> i64 {
1004 self.next_id.fetch_add(1, Ordering::Relaxed)
1005 }
1006 }
1007
1008 /// Flatten an MCP `tools/call` result into text. Joins text content blocks;
1009 /// notes non-text blocks; honors `isError`.
1010 fn render_tool_result(result: &Value) -> String {
1011 let mut out = String::new();
1012 if let Some(blocks) = result.get("content").and_then(Value::as_array) {
1013 for block in blocks {
1014 match block.get("type").and_then(Value::as_str) {
1015 Some("text") => {
1016 if let Some(text) = block.get("text").and_then(Value::as_str) {
1017 if !out.is_empty() {
1018 out.push('\n');
1019 }
1020 out.push_str(text);
1021 }
1022 }
1023 Some(other) => {
1024 if !out.is_empty() {
1025 out.push('\n');
1026 }
1027 out.push_str(&format!("[{other} content omitted]"));
1028 }
1029 None => {}
1030 }
1031 }
1032 }
1033
1034 // Some servers return only `structuredContent`; surface it if there was no
1035 // textual content.
1036 if out.is_empty()
1037 && let Some(structured) = result.get("structuredContent")
1038 {
1039 out = structured.to_string();
1040 }
1041
1042 if out.is_empty() {
1043 out = "(tool returned no content)".to_string();
1044 }
1045
1046 if result.get("isError").and_then(Value::as_bool) == Some(true) {
1047 format!("Tool reported an error:\n{out}")
1048 } else {
1049 out
1050 }
1051 }
1052
1053 /// Truncate tool output to the context-protecting limit, with a trailing note.
1054 fn truncate(text: String) -> String {
1055 if text.chars().count() <= RESULT_CHAR_LIMIT {
1056 return text;
1057 }
1058 let kept: String = text.chars().take(RESULT_CHAR_LIMIT).collect();
1059 format!("{kept}\n\n[output truncated to {RESULT_CHAR_LIMIT} characters]")
1060 }
1061
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
1478 // ── Status reporting (`/mcp`) ────────────────────────────────────────────────
1479
1480 /// Human-readable summary of configured MCP servers and their tools, for the
1481 /// `/mcp` slash command. Shows the URL for HTTP servers and the command line
1482 /// for stdio servers.
1483 pub fn status_summary() -> String {
1484 let session = session_servers();
1485 let Some(mcp) = MCP.get() else {
1486 if session.0.is_empty() {
1487 return "MCP is not initialized.".to_string();
1488 }
1489 return summarize_servers(&[], &session.0);
1490 };
1491 if mcp.servers.is_empty() && session.0.is_empty() {
1492 return "No MCP servers configured. Add one in ~/.config/sigit/mcp.toml \
1493 or .sigit/mcp.toml. See https://modelcontextprotocol.io."
1494 .to_string();
1495 }
1496 summarize_servers(&mcp.servers, &session.0)
1497 }
1498
1499 /// The `/mcp` listing for the startup servers followed by the ones the live
1500 /// session's client supplied, which are marked as such.
1501 fn summarize_servers(startup: &[ServerConn], session: &[ServerConn]) -> String {
1502 let all = || startup.iter().chain(session.iter());
1503 let total_tools: usize = all().map(|s| s.tools.len()).sum();
1504 let mut lines = vec![format!(
1505 "{} MCP server(s), {total_tools} tool(s) available:",
1506 all().count()
1507 )];
1508 for (index, server) in all().enumerate() {
1509 let origin = if index >= startup.len() {
1510 ", from the editor"
1511 } else {
1512 ""
1513 };
1514 match &server.error {
1515 Some(error) => lines.push(format!(
1516 "- {} ({}{origin}) — unavailable: {error}",
1517 server.name, server.endpoint
1518 )),
1519 None => {
1520 lines.push(format!(
1521 "- {} ({}{origin}) — {} tool(s)",
1522 server.name,
1523 server.endpoint,
1524 server.tools.len()
1525 ));
1526 for tool in &server.tools {
1527 lines.push(format!(" • {}", tool.full_name));
1528 }
1529 }
1530 }
1531 }
1532 lines.join("\n")
1533 }
1534
1535 #[cfg(test)]
1536 mod tests {
1537 use super::*;
1538
1539 #[test]
1540 fn is_mcp_tool_detects_prefix() {
1541 assert!(is_mcp_tool("mcp__sigit__search"));
1542 assert!(!is_mcp_tool("read_file"));
1543 assert!(!is_mcp_tool("skill"));
1544 }
1545
1546 #[test]
1547 fn official_tool_name_matches_the_namespacing_convention() {
1548 assert_eq!(official_tool_name("list_issues"), "mcp__sigit__list_issues");
1549 assert_eq!(
1550 official_tool_name("get_pull_request"),
1551 "mcp__sigit__get_pull_request"
1552 );
1553 }
1554
1555 #[test]
1556 fn official_tool_suffix_strips_only_the_official_namespace() {
1557 assert_eq!(
1558 official_tool_suffix("mcp__sigit__list_issues"),
1559 Some("list_issues")
1560 );
1561 assert_eq!(official_tool_suffix("mcp__other__list_issues"), None);
1562 // `sigit` must be the whole server name, not a prefix of it.
1563 assert_eq!(official_tool_suffix("mcp__sigitx__list_issues"), None);
1564 assert_eq!(official_tool_suffix("list_issues"), None);
1565 assert_eq!(official_tool_suffix("mcp__sigit__"), Some(""));
1566 }
1567
1568 #[test]
1569 fn smbcloud_tool_suffix_strips_only_the_smbcloud_namespace() {
1570 assert_eq!(
1571 smbcloud_tool_suffix("mcp__smbcloud__project_list"),
1572 Some("project_list")
1573 );
1574 assert_eq!(smbcloud_tool_suffix("mcp__sigit__project_list"), None);
1575 // `smbcloud` must be the whole server name, not a prefix of it.
1576 assert_eq!(smbcloud_tool_suffix("mcp__smbcloudx__project_list"), None);
1577 assert_eq!(smbcloud_tool_suffix("project_list"), None);
1578 assert_eq!(smbcloud_tool_suffix("mcp__smbcloud__"), Some(""));
1579 }
1580
1581 #[test]
1582 fn sanitize_collapses_invalid_chars() {
1583 assert_eq!(sanitize("github"), "github");
1584 assert_eq!(sanitize("my server"), "my_server");
1585 assert_eq!(sanitize("a.b/c:d"), "a_b_c_d");
1586 assert_eq!(sanitize("keep-_ok9"), "keep-_ok9");
1587 }
1588
1589 #[test]
1590 fn parses_mcp_file_with_servers() {
1591 let toml = r#"
1592 official = false
1593
1594 [[server]]
1595 name = "github"
1596 url = "https://api.example.com/mcp"
1597
1598 [[server]]
1599 name = "disabled-one"
1600 url = "https://nope.example.com/mcp"
1601 enabled = false
1602
1603 [server.headers]
1604 Authorization = "Bearer xyz"
1605 "#;
1606 let parsed: McpFile = toml::from_str(toml).unwrap();
1607 assert_eq!(parsed.official, Some(false));
1608 assert_eq!(parsed.smbcloud, None);
1609 assert_eq!(parsed.server.len(), 2);
1610 assert_eq!(parsed.server[0].name, "github");
1611 assert_eq!(parsed.server[1].enabled, Some(false));
1612 assert_eq!(
1613 parsed.server[1]
1614 .headers
1615 .get("Authorization")
1616 .map(String::as_str),
1617 Some("Bearer xyz")
1618 );
1619 }
1620
1621 #[test]
1622 fn parses_smbcloud_opt_out_flag() {
1623 let parsed: McpFile = toml::from_str("smbcloud = false").unwrap();
1624 assert_eq!(parsed.smbcloud, Some(false));
1625 // Absent means "include" (the default stays true in load_configs).
1626 let parsed: McpFile = toml::from_str("").unwrap();
1627 assert_eq!(parsed.smbcloud, None);
1628 }
1629
1630 #[test]
1631 fn on_path_finds_real_binaries_only() {
1632 assert!(!on_path("definitely-not-a-real-binary-xyzzy"));
1633 #[cfg(unix)]
1634 assert!(on_path("sh"));
1635 }
1636
1637 #[test]
1638 fn parses_stdio_server_with_args_and_env() {
1639 let toml = r#"
1640 [[server]]
1641 name = "fs"
1642 command = "npx"
1643 args = ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"]
1644
1645 [server.env]
1646 LOG_LEVEL = "debug"
1647 TOKEN = "abc"
1648 "#;
1649 let parsed: McpFile = toml::from_str(toml).unwrap();
1650 assert_eq!(parsed.server.len(), 1);
1651 let entry = &parsed.server[0];
1652 assert_eq!(entry.command.as_deref(), Some("npx"));
1653 assert_eq!(entry.args.len(), 3);
1654 assert_eq!(
1655 entry.env.get("LOG_LEVEL").map(String::as_str),
1656 Some("debug")
1657 );
1658 assert_eq!(entry.env.get("TOKEN").map(String::as_str), Some("abc"));
1659
1660 let def = entry.transport_def().expect("valid stdio entry");
1661 match def {
1662 TransportDef::Stdio { command, args, env } => {
1663 assert_eq!(command, "npx");
1664 assert_eq!(args[0], "-y");
1665 assert_eq!(env.len(), 2);
1666 }
1667 TransportDef::Http { .. } => panic!("expected a stdio transport"),
1668 }
1669 }
1670
1671 #[test]
1672 fn entry_with_url_and_command_is_a_config_error() {
1673 let toml = r#"
1674 [[server]]
1675 name = "confused"
1676 url = "https://example.com/mcp"
1677 command = "npx"
1678 "#;
1679 let parsed: McpFile = toml::from_str(toml).unwrap();
1680 let error = parsed.server[0].transport_def().unwrap_err();
1681 assert!(error.contains("both"), "unexpected error: {error}");
1682 }
1683
1684 #[test]
1685 fn entry_with_neither_url_nor_command_is_a_config_error() {
1686 let toml = r#"
1687 [[server]]
1688 name = "empty"
1689 "#;
1690 let parsed: McpFile = toml::from_str(toml).unwrap();
1691 let error = parsed.server[0].transport_def().unwrap_err();
1692 assert!(error.contains("needs"), "unexpected error: {error}");
1693 }
1694
1695 #[test]
1696 fn blank_url_or_command_counts_as_absent() {
1697 let toml = r#"
1698 [[server]]
1699 name = "blank"
1700 url = " "
1701 command = "server-bin"
1702 "#;
1703 let parsed: McpFile = toml::from_str(toml).unwrap();
1704 // A blank url is treated as absent, so this resolves to stdio.
1705 match parsed.server[0].transport_def().expect("stdio") {
1706 TransportDef::Stdio { command, .. } => assert_eq!(command, "server-bin"),
1707 TransportDef::Http { .. } => panic!("expected stdio"),
1708 }
1709 }
1710
1711 #[test]
1712 fn endpoint_renders_url_or_command_line() {
1713 let http = TransportDef::Http {
1714 url: "https://example.com/mcp".into(),
1715 headers: vec![],
1716 };
1717 assert_eq!(http.endpoint(), "https://example.com/mcp");
1718
1719 let stdio = TransportDef::Stdio {
1720 command: "npx".into(),
1721 args: vec!["-y".into(), "server-fs".into()],
1722 env: vec![],
1723 };
1724 assert_eq!(stdio.endpoint(), "npx -y server-fs");
1725 }
1726
1727 #[test]
1728 fn upsert_replaces_same_name() {
1729 let mut defs = vec![ServerDef {
1730 name: "a".into(),
1731 transport: TransportDef::Http {
1732 url: "u1".into(),
1733 headers: vec![],
1734 },
1735 }];
1736 upsert(
1737 &mut defs,
1738 ServerDef {
1739 name: "a".into(),
1740 transport: TransportDef::Http {
1741 url: "u2".into(),
1742 headers: vec![],
1743 },
1744 },
1745 );
1746 assert_eq!(defs.len(), 1);
1747 assert_eq!(defs[0].transport.endpoint(), "u2");
1748 }
1749
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
1771 #[test]
1772 fn render_result_joins_text_blocks() {
1773 let result = json!({
1774 "content": [
1775 { "type": "text", "text": "line one" },
1776 { "type": "text", "text": "line two" }
1777 ]
1778 });
1779 assert_eq!(render_tool_result(&result), "line one\nline two");
1780 }
1781
1782 #[test]
1783 fn render_result_marks_errors_and_non_text() {
1784 let result = json!({
1785 "isError": true,
1786 "content": [
1787 { "type": "text", "text": "boom" },
1788 { "type": "image", "data": "..." }
1789 ]
1790 });
1791 let rendered = render_tool_result(&result);
1792 assert!(rendered.starts_with("Tool reported an error:"));
1793 assert!(rendered.contains("boom"));
1794 assert!(rendered.contains("[image content omitted]"));
1795 }
1796
1797 #[test]
1798 fn render_result_falls_back_to_structured_content() {
1799 let result = json!({ "structuredContent": { "value": 42 } });
1800 assert!(render_tool_result(&result).contains("42"));
1801 }
1802
1803 #[test]
1804 fn truncate_caps_long_output() {
1805 let long = "x".repeat(RESULT_CHAR_LIMIT + 100);
1806 let out = truncate(long);
1807 assert!(out.contains("[output truncated"));
1808 }
1809
1810 #[test]
1811 fn tool_specs_empty_before_init() {
1812 // Without init() the global is unset; this must not panic.
1813 assert!(super::tool_specs().is_empty() || MCP.get().is_some());
1814 }
1815 }