@setoelkahfi / sigit
Code Commits Pulls CI/CD 248
development
claude/code-feature-parity-q003hm claude/elegant-carson-l1menh claude/sigit-acp-local-chat-cx6380 claude/sigit-cloud-agent-expansion-reox0a claude/tool-permission-system claude/zen-feynman-0u78dk development feature/agent-tools-multiedit-glob-todos-remember feature/background-commands feature/commit-coauthor-attribution feature/headless-mode feature/init-command feature/load-local-model-explicitly feature/session-persistence-compaction feature/sigit-code-cloud feature/subagent-tool feature/tool-permission-system feature/tui-repo-tabs feature/tui-tabs main release/v1.3.1
rs 1,486 lines 51.8 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 //! ## The official server
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. Additional servers are configured in `mcp.toml` (see
30 //! [`load_configs`]).
31 //!
32 //! ## Lifecycle
33 //!
34 //! Discovery is best-effort and happens once at startup via [`init`]: each
35 //! configured server is contacted concurrently (with a per-server timeout),
36 //! runs the `initialize` handshake, and has its `tools/list` cached. A server
37 //! that fails to connect is recorded with its error and simply contributes no
38 //! tools — it never blocks startup or the rest of the agent. The result is
39 //! stored in a process-global so the synchronous tool-spec builders
40 //! ([`tool_specs`]) and the async dispatch ([`call_tool`]) can both read it.
41 //!
42 //! stdio children live for the sigit process. When a child dies (EOF or an I/O
43 //! error on its pipes) the server is marked dead and later calls return an
44 //! in-band error string the model can react to; there is no automatic restart.
45 //! `/reload` does *not* re-run discovery ([`init`] is once-per-process), so a
46 //! changed `mcp.toml` or a dead server needs a sigit restart. At process exit
47 //! children see EOF on their stdin and exit on their own.
48 //!
49 //! Tools are namespaced `mcp__<server>__<tool>` so they never collide with
50 //! built-in tools or with each other across servers. This mirrors the
51 //! convention used by other MCP-aware agents.
52 //!
53 //! Like the rest of the backend seam, MCP is wired up only through the
54 //! interactive client and the ACP agent loop. On non-Unix targets a few helpers
55 //! are unused, so the dead-code lint is suppressed there only.
56 #![cfg_attr(not(unix), allow(dead_code))]
57
58 use std::collections::{BTreeMap, HashMap};
59 use std::path::PathBuf;
60 use std::sync::Mutex as StdMutex;
61 use std::sync::atomic::{AtomicI64, Ordering};
62 use std::sync::{Arc, OnceLock};
63 use std::time::Duration;
64
65 use serde::Deserialize;
66 use serde_json::{Value, json};
67 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
68 use tokio::process::{Child, ChildStdin, ChildStdout, Command};
69 use tokio::sync::{Mutex, oneshot};
70
71 use crate::backend::ToolSpec;
72
73 /// Prefix marking a tool as MCP-provided. The full name is
74 /// `mcp__<server>__<tool>`.
75 pub const MCP_PREFIX: &str = "mcp__";
76
77 /// JSON-RPC / MCP protocol version we advertise in the handshake.
78 const PROTOCOL_VERSION: &str = "2025-06-18";
79
80 /// Per-server budget for the connect + `initialize` + `tools/list` handshake at
81 /// startup. Bounds how long an unreachable server can delay startup; servers are
82 /// contacted concurrently, so this is the worst case for the whole set, not the
83 /// sum.
84 const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(8);
85
86 /// Overall request timeout for an individual `tools/call`. Generous, since an
87 /// MCP tool may do real work server-side.
88 const CALL_TIMEOUT: Duration = Duration::from_secs(120);
89
90 /// Cap on the characters returned from a single tool call, so a chatty server
91 /// can't blow up the model's context. Matches the spirit of the file-read cap.
92 const RESULT_CHAR_LIMIT: usize = 30_000;
93
94 // ── Public types ──────────────────────────────────────────────────────────────
95
96 /// A tool discovered on an MCP server, in siGit's flattened form.
97 #[derive(Debug, Clone)]
98 struct McpTool {
99 /// Namespaced name exposed to the model: `mcp__<server>__<tool>`.
100 full_name: String,
101 /// The tool's name as the server knows it (sent back in `tools/call`).
102 remote_name: String,
103 /// Human/model-facing description, prefixed with the server name.
104 description: String,
105 /// JSON Schema for the tool's arguments, encoded as a string.
106 parameters_schema: String,
107 }
108
109 /// A configured MCP server and its live connection state.
110 struct ServerConn {
111 /// Sanitized server name used in tool namespacing and the `/mcp` listing.
112 name: String,
113 /// Display endpoint for the `/mcp` listing: the URL for HTTP servers, the
114 /// command line for stdio servers.
115 endpoint: String,
116 /// The live transport. `None` when a stdio server failed to even spawn.
117 transport: Option<Transport>,
118 /// Tools discovered at startup. Empty when the server failed to connect.
119 tools: Vec<McpTool>,
120 /// Connection error, if the handshake failed. Surfaced by `/mcp`.
121 error: Option<String>,
122 }
123
124 /// How a connected server is reached.
125 enum Transport {
126 Http(HttpConn),
127 Stdio(StdioConn),
128 }
129
130 /// Streamable HTTP connection state.
131 struct HttpConn {
132 /// Streamable HTTP endpoint (the single POST URL).
133 url: String,
134 /// Extra headers sent on every request (e.g. `Authorization`).
135 headers: Vec<(String, String)>,
136 /// Session id handed back by the server on `initialize`, echoed on every
137 /// later request via the `Mcp-Session-Id` header.
138 session_id: Mutex<Option<String>>,
139 }
140
141 /// stdio connection state: a child process speaking newline-delimited JSON-RPC
142 /// over its stdin/stdout.
143 struct StdioConn {
144 /// The child's stdin. The mutex serializes writes so concurrent requests
145 /// can't interleave bytes on the pipe; `None` once the pipe broke.
146 writer: Mutex<Option<ChildStdin>>,
147 /// State shared with the background reader task that owns the child's
148 /// stdout.
149 shared: Arc<StdioShared>,
150 /// JSON-RPC id source. Ids are per-connection so the reader task can route
151 /// each response to the request that carries its id.
152 next_id: AtomicI64,
153 }
154
155 /// State shared between a [`StdioConn`] and its background reader task.
156 struct StdioShared {
157 /// Server name, for log lines.
158 name: String,
159 /// In-flight requests awaiting a response, keyed by JSON-RPC id. Dropping
160 /// a sender (when the connection dies) wakes the waiter with an error.
161 pending: StdMutex<HashMap<i64, oneshot::Sender<Value>>>,
162 /// Why the connection is unusable, once it is (EOF, I/O error, kill).
163 dead: StdMutex<Option<String>>,
164 /// The child handle, kept so a dead/failed connection can kill and reap
165 /// the process. Taken on death.
166 child: StdMutex<Option<Child>>,
167 }
168
169 impl StdioShared {
170 fn dead_reason(&self) -> Option<String> {
171 self.dead.lock().unwrap().clone()
172 }
173
174 /// Mark the connection unusable: record the reason (first one wins), fail
175 /// every in-flight request, and kill + reap the child, best effort.
176 fn mark_dead(&self, reason: &str) {
177 {
178 let mut dead = self.dead.lock().unwrap();
179 if dead.is_none() {
180 *dead = Some(reason.to_string());
181 }
182 }
183 // Dropping the senders wakes every waiter with a recv error.
184 self.pending.lock().unwrap().clear();
185 if let Some(mut child) = self.child.lock().unwrap().take() {
186 let _ = child.start_kill();
187 tokio::spawn(async move {
188 let _ = child.wait().await;
189 });
190 }
191 }
192 }
193
194 /// The process-global MCP state: a shared HTTP client plus every configured
195 /// server.
196 struct Mcp {
197 http: reqwest::Client,
198 servers: Vec<ServerConn>,
199 next_id: AtomicI64,
200 }
201
202 static MCP: OnceLock<Mcp> = OnceLock::new();
203
204 // ── Configuration ───────────────────────────────────────────────────────────
205
206 /// Default endpoint of the official siGit Code MCP server, derived from the
207 /// cloud base URL so `SIGIT_CLOUD_URL` (dev) carries over.
208 fn official_url() -> String {
209 format!(
210 "{}/mcp",
211 crate::provider::cloud_base_url().trim_end_matches('/')
212 )
213 }
214
215 /// A server entry as written in `mcp.toml`. Exactly one of `url` (Streamable
216 /// HTTP) or `command` (stdio) selects the transport.
217 #[derive(Debug, Deserialize)]
218 struct ServerEntry {
219 name: String,
220 /// Streamable HTTP endpoint. Mutually exclusive with `command`.
221 #[serde(default)]
222 url: Option<String>,
223 /// stdio server executable. Mutually exclusive with `url`.
224 #[serde(default)]
225 command: Option<String>,
226 /// Arguments for `command`.
227 #[serde(default)]
228 args: Vec<String>,
229 /// Extra environment variables for `command`, added on top of the
230 /// inherited environment.
231 #[serde(default)]
232 env: BTreeMap<String, String>,
233 /// Set `enabled = false` to keep an entry in the file but skip connecting.
234 #[serde(default)]
235 enabled: Option<bool>,
236 /// Static headers, e.g. `Authorization = "Bearer ..."`. HTTP only.
237 #[serde(default)]
238 headers: BTreeMap<String, String>,
239 }
240
241 impl ServerEntry {
242 /// Resolve the entry's transport. `url` and `command` are mutually
243 /// exclusive and exactly one is required; anything else is a config error.
244 fn transport_def(&self) -> Result<TransportDef, String> {
245 let url = self.url.as_deref().map(str::trim).filter(|v| !v.is_empty());
246 let command = self
247 .command
248 .as_deref()
249 .map(str::trim)
250 .filter(|v| !v.is_empty());
251 match (url, command) {
252 (Some(_), Some(_)) => {
253 Err("has both `url` and `command`; a server uses exactly one transport".to_string())
254 }
255 (None, None) => {
256 Err("needs either `url` (Streamable HTTP) or `command` (stdio)".to_string())
257 }
258 (Some(url), None) => Ok(TransportDef::Http {
259 url: url.to_string(),
260 headers: self
261 .headers
262 .iter()
263 .map(|(k, v)| (k.clone(), v.clone()))
264 .collect(),
265 }),
266 (None, Some(command)) => Ok(TransportDef::Stdio {
267 command: command.to_string(),
268 args: self.args.clone(),
269 env: self
270 .env
271 .iter()
272 .map(|(k, v)| (k.clone(), v.clone()))
273 .collect(),
274 }),
275 }
276 }
277 }
278
279 /// The `mcp.toml` schema.
280 #[derive(Debug, Default, Deserialize)]
281 struct McpFile {
282 /// Include the baked-in official server. Defaults to `true`; set `false` to
283 /// opt out.
284 #[serde(default)]
285 official: Option<bool>,
286 #[serde(default)]
287 server: Vec<ServerEntry>,
288 }
289
290 /// How to reach a configured server, before connecting.
291 #[derive(Debug, Clone)]
292 enum TransportDef {
293 Http {
294 url: String,
295 headers: Vec<(String, String)>,
296 },
297 Stdio {
298 command: String,
299 args: Vec<String>,
300 env: Vec<(String, String)>,
301 },
302 }
303
304 impl TransportDef {
305 /// Human-readable endpoint for logs and the `/mcp` listing: the URL for
306 /// HTTP, the command line for stdio.
307 fn endpoint(&self) -> String {
308 match self {
309 TransportDef::Http { url, .. } => url.clone(),
310 TransportDef::Stdio { command, args, .. } => {
311 let mut line = command.clone();
312 for arg in args {
313 line.push(' ');
314 line.push_str(arg);
315 }
316 line
317 }
318 }
319 }
320 }
321
322 /// A resolved server definition, before connecting.
323 #[derive(Debug, Clone)]
324 struct ServerDef {
325 name: String,
326 transport: TransportDef,
327 }
328
329 /// Config files to read, in priority order (later wins on a name clash):
330 /// global `$SIGIT_CONFIG_DIR/mcp.toml`, then project-local `<cwd>/.sigit/mcp.toml`.
331 fn config_paths() -> Vec<PathBuf> {
332 let mut paths = Vec::new();
333 if let Some(dir) = sigit_config_dir() {
334 paths.push(dir.join("mcp.toml"));
335 }
336 if let Ok(cwd) = std::env::current_dir() {
337 paths.push(cwd.join(".sigit").join("mcp.toml"));
338 }
339 paths
340 }
341
342 fn sigit_config_dir() -> Option<PathBuf> {
343 if let Ok(dir) = std::env::var("SIGIT_CONFIG_DIR")
344 && !dir.is_empty()
345 {
346 return Some(PathBuf::from(dir));
347 }
348 std::env::var("HOME")
349 .ok()
350 .map(|home| PathBuf::from(home).join(".config").join("sigit"))
351 }
352
353 /// Resolve the full set of servers to connect to: the baked-in official server
354 /// (unless opted out) plus any from `mcp.toml`. Project-local entries override
355 /// global ones, and a user entry named `sigit` overrides the official default.
356 fn load_configs() -> Vec<ServerDef> {
357 // Global escape hatch: `SIGIT_MCP=off` disables MCP entirely.
358 if let Ok(value) = std::env::var("SIGIT_MCP")
359 && matches!(
360 value.trim().to_ascii_lowercase().as_str(),
361 "off" | "0" | "false" | "no" | "disabled"
362 )
363 {
364 log::info!("mcp: disabled via SIGIT_MCP");
365 return Vec::new();
366 }
367
368 let mut include_official = true;
369 // De-duplicated by sanitized name; a later config file overrides an earlier
370 // one for the same name (project-local wins over global).
371 let mut defs: Vec<ServerDef> = Vec::new();
372
373 for path in config_paths() {
374 let Ok(contents) = std::fs::read_to_string(&path) else {
375 continue;
376 };
377 let parsed: McpFile = match toml::from_str(&contents) {
378 Ok(parsed) => parsed,
379 Err(error) => {
380 log::warn!("mcp: ignoring {}: {error}", path.display());
381 continue;
382 }
383 };
384 if let Some(official) = parsed.official {
385 include_official = official;
386 }
387 for entry in parsed.server {
388 if entry.enabled == Some(false) {
389 continue;
390 }
391 let name = sanitize(&entry.name);
392 if name.is_empty() {
393 log::warn!("mcp: skipping server with empty name in {}", path.display());
394 continue;
395 }
396 let transport = match entry.transport_def() {
397 Ok(transport) => transport,
398 Err(error) => {
399 log::warn!(
400 "mcp: skipping server '{name}' in {}: {error}",
401 path.display()
402 );
403 continue;
404 }
405 };
406 upsert(&mut defs, ServerDef { name, transport });
407 }
408 }
409
410 // The official server can also be disabled with SIGIT_MCP_OFFICIAL=off.
411 if let Ok(value) = std::env::var("SIGIT_MCP_OFFICIAL")
412 && matches!(
413 value.trim().to_ascii_lowercase().as_str(),
414 "off" | "0" | "false" | "no"
415 )
416 {
417 include_official = false;
418 }
419
420 // Add the baked-in official server, but never clobber a user-defined entry
421 // named `sigit` — an explicit config (e.g. a custom URL or headers) wins.
422 if include_official && !defs.iter().any(|d| d.name == "sigit") {
423 let mut headers = Vec::new();
424 if let Some(token) = crate::credentials::load_token() {
425 headers.push(("Authorization".to_string(), format!("Bearer {token}")));
426 }
427 defs.push(ServerDef {
428 name: "sigit".to_string(),
429 transport: TransportDef::Http {
430 url: official_url(),
431 headers,
432 },
433 });
434 }
435
436 defs
437 }
438
439 /// Insert `def`, replacing any existing entry with the same name.
440 fn upsert(defs: &mut Vec<ServerDef>, def: ServerDef) {
441 if let Some(slot) = defs.iter_mut().find(|d| d.name == def.name) {
442 *slot = def;
443 } else {
444 defs.push(def);
445 }
446 }
447
448 /// Sanitize a name into the `[a-zA-Z0-9_-]` set tool names are restricted to,
449 /// collapsing anything else to `_`.
450 fn sanitize(raw: &str) -> String {
451 raw.trim()
452 .chars()
453 .map(|c| {
454 if c.is_ascii_alphanumeric() || c == '_' || c == '-' {
455 c
456 } else {
457 '_'
458 }
459 })
460 .collect()
461 }
462
463 // ── Startup / discovery ─────────────────────────────────────────────────────
464
465 /// Connect to every configured server and cache the tools they expose. Idempotent
466 /// and best-effort: a server that can't be reached is recorded with its error and
467 /// contributes no tools. Safe to call from either entry point; only the first
468 /// call does work.
469 pub async fn init() {
470 if MCP.get().is_some() {
471 return;
472 }
473
474 let defs = load_configs();
475 let http = reqwest::Client::builder()
476 .timeout(CALL_TIMEOUT)
477 .user_agent(concat!(
478 "sigit/",
479 env!("CARGO_PKG_VERSION"),
480 " (mcp-client)"
481 ))
482 .build()
483 .unwrap_or_default();
484
485 // Contact servers concurrently so one slow/unreachable host doesn't serialize
486 // the rest. Each handshake is bounded by HANDSHAKE_TIMEOUT.
487 let connects = defs.into_iter().map(|def| {
488 let http = http.clone();
489 async move { connect(&http, def).await }
490 });
491 let servers = futures::future::join_all(connects).await;
492
493 for server in &servers {
494 match &server.error {
495 Some(error) => log::warn!("mcp: server '{}' unavailable: {error}", server.name),
496 None => log::info!(
497 "mcp: server '{}' ready, {} tool(s)",
498 server.name,
499 server.tools.len()
500 ),
501 }
502 }
503
504 let _ = MCP.set(Mcp {
505 http,
506 servers,
507 next_id: AtomicI64::new(1),
508 });
509 }
510
511 /// Run the handshake against one server and collect its tools. Always returns a
512 /// `ServerConn`; failures land in its `error` field rather than propagating.
513 async fn connect(http: &reqwest::Client, def: ServerDef) -> ServerConn {
514 let endpoint = def.transport.endpoint();
515 let transport = match &def.transport {
516 TransportDef::Http { url, headers } => Transport::Http(HttpConn {
517 url: url.clone(),
518 headers: headers.clone(),
519 session_id: Mutex::new(None),
520 }),
521 TransportDef::Stdio { command, args, env } => {
522 match spawn_stdio(&def.name, command, args, env) {
523 Ok(conn) => Transport::Stdio(conn),
524 Err(error) => {
525 return ServerConn {
526 name: def.name,
527 endpoint,
528 transport: None,
529 tools: Vec::new(),
530 error: Some(error),
531 };
532 }
533 }
534 }
535 };
536
537 let mut conn = ServerConn {
538 name: def.name,
539 endpoint,
540 transport: Some(transport),
541 tools: Vec::new(),
542 error: None,
543 };
544
545 let handshake = tokio::time::timeout(HANDSHAKE_TIMEOUT, async {
546 // initialize → notifications/initialized → tools/list
547 initialize(http, &conn).await?;
548 notify_initialized(http, &conn).await?;
549 list_tools(http, &conn).await
550 })
551 .await;
552
553 match handshake {
554 Ok(Ok(tools)) => conn.tools = tools,
555 Ok(Err(error)) => conn.error = Some(error),
556 Err(_) => conn.error = Some(format!("timed out after {}s", HANDSHAKE_TIMEOUT.as_secs())),
557 }
558
559 // A stdio child that failed its handshake is useless — kill it rather than
560 // leave it running for the rest of the process.
561 if let Some(error) = conn.error.clone()
562 && let Some(Transport::Stdio(stdio)) = &conn.transport
563 {
564 stdio.shared.mark_dead(&error);
565 }
566
567 conn
568 }
569
570 /// The `initialize` request: negotiate protocol version and (on HTTP) capture
571 /// the session id from the response headers (handled inside [`post_rpc`]).
572 async fn initialize(http: &reqwest::Client, conn: &ServerConn) -> Result<(), String> {
573 let params = json!({
574 "protocolVersion": PROTOCOL_VERSION,
575 "capabilities": {},
576 "clientInfo": { "name": "sigit", "version": env!("CARGO_PKG_VERSION") }
577 });
578 rpc_request(http, conn, "initialize", params, HANDSHAKE_TIMEOUT).await?;
579 Ok(())
580 }
581
582 /// The `notifications/initialized` notification. Servers expect it before
583 /// fielding requests; it carries no id and no response.
584 async fn notify_initialized(http: &reqwest::Client, conn: &ServerConn) -> Result<(), String> {
585 rpc_notify(http, conn, "notifications/initialized", HANDSHAKE_TIMEOUT).await
586 }
587
588 /// `tools/list`, following `nextCursor` pagination, mapped into [`McpTool`]s.
589 async fn list_tools(http: &reqwest::Client, conn: &ServerConn) -> Result<Vec<McpTool>, String> {
590 let mut tools = Vec::new();
591 let mut cursor: Option<String> = None;
592
593 loop {
594 let params = match &cursor {
595 Some(c) => json!({ "cursor": c }),
596 None => json!({}),
597 };
598 let result = rpc_request(http, conn, "tools/list", params, HANDSHAKE_TIMEOUT).await?;
599
600 for tool in result
601 .get("tools")
602 .and_then(Value::as_array)
603 .into_iter()
604 .flatten()
605 {
606 let Some(remote_name) = tool.get("name").and_then(Value::as_str) else {
607 continue;
608 };
609 let full_name = format!("{MCP_PREFIX}{}__{}", conn.name, sanitize(remote_name));
610 if full_name.chars().count() > 64 {
611 log::warn!(
612 "mcp: tool name '{full_name}' exceeds 64 chars; some backends may reject it"
613 );
614 }
615 let remote_desc = tool
616 .get("description")
617 .and_then(Value::as_str)
618 .unwrap_or("")
619 .trim();
620 let description = if remote_desc.is_empty() {
621 format!("[MCP server '{}'] {remote_name}", conn.name)
622 } else {
623 format!("[MCP server '{}'] {remote_desc}", conn.name)
624 };
625 // `inputSchema` is a JSON Schema object; default to a permissive
626 // object schema when a server omits it.
627 let parameters_schema = tool
628 .get("inputSchema")
629 .filter(|schema| schema.is_object())
630 .cloned()
631 .unwrap_or_else(|| json!({ "type": "object" }))
632 .to_string();
633
634 tools.push(McpTool {
635 full_name,
636 remote_name: remote_name.to_string(),
637 description,
638 parameters_schema,
639 });
640 }
641
642 cursor = result
643 .get("nextCursor")
644 .and_then(Value::as_str)
645 .map(str::to_string);
646 if cursor.is_none() {
647 break;
648 }
649 }
650
651 Ok(tools)
652 }
653
654 // ── Tool exposure + dispatch ────────────────────────────────────────────────
655
656 /// Whether a tool name belongs to MCP. The dispatch in `tools::execute_tool`
657 /// uses this to route a call here.
658 pub fn is_mcp_tool(name: &str) -> bool {
659 name.starts_with(MCP_PREFIX)
660 }
661
662 /// All discovered MCP tools as agent [`ToolSpec`]s, ready to append to the
663 /// built-in tool list. Empty when MCP is uninitialized or no server exposed any.
664 pub fn tool_specs() -> Vec<ToolSpec> {
665 let Some(mcp) = MCP.get() else {
666 return Vec::new();
667 };
668 let mut specs = Vec::new();
669 for server in &mcp.servers {
670 for tool in &server.tools {
671 specs.push(ToolSpec {
672 name: tool.full_name.clone(),
673 description: tool.description.clone(),
674 parameters_schema: tool.parameters_schema.clone(),
675 });
676 }
677 }
678 specs
679 }
680
681 /// Execute an MCP tool call by name, returning text to feed back to the model.
682 /// Errors are returned as plain strings (never panics) so a failing tool degrades
683 /// to a message the model can react to, exactly like the built-in tools.
684 pub async fn call_tool(full_name: &str, arguments: &str) -> String {
685 let Some(mcp) = MCP.get() else {
686 return "Error: MCP is not initialized.".to_string();
687 };
688
689 let Some((server, tool)) = mcp.servers.iter().find_map(|s| {
690 s.tools
691 .iter()
692 .find(|t| t.full_name == full_name)
693 .map(|t| (s, t))
694 }) else {
695 return format!("Error: unknown MCP tool \"{full_name}\".");
696 };
697
698 // Arguments arrive as a JSON-encoded string; an empty/blank string means no
699 // arguments. Anything that isn't a JSON object is a model mistake.
700 let args: Value = if arguments.trim().is_empty() {
701 json!({})
702 } else {
703 match serde_json::from_str(arguments) {
704 Ok(value @ Value::Object(_)) => value,
705 Ok(_) => return "Error: tool arguments must be a JSON object.".to_string(),
706 Err(error) => return format!("Error: failed to parse arguments: {error}"),
707 }
708 };
709
710 match mcp.call(server, &tool.remote_name, args).await {
711 Ok(text) => truncate(text),
712 Err(error) => format!("Error: {error}"),
713 }
714 }
715
716 impl Mcp {
717 /// Send a `tools/call` and render the result into text. On HTTP, retries
718 /// once after a re-`initialize` if the session was dropped (HTTP 404),
719 /// which is how Streamable HTTP signals an expired session.
720 async fn call(
721 &self,
722 server: &ServerConn,
723 remote_name: &str,
724 args: Value,
725 ) -> Result<String, String> {
726 let params = json!({ "name": remote_name, "arguments": args });
727 let result = match &server.transport {
728 None => return Err(format!("server '{}' is not connected", server.name)),
729 Some(Transport::Stdio(stdio)) => {
730 stdio.request("tools/call", params, CALL_TIMEOUT).await?
731 }
732 Some(Transport::Http(http_conn)) => {
733 let body = json!({
734 "jsonrpc": "2.0",
735 "id": 0,
736 "method": "tools/call",
737 "params": params
738 });
739 match post_rpc(&self.http, &server.name, http_conn, &body, CALL_TIMEOUT).await {
740 Ok(result) => result,
741 Err(error) if error.contains("returned 404") => {
742 // Session expired — drop it, re-handshake, and retry once.
743 *http_conn.session_id.lock().await = None;
744 initialize(&self.http, server).await?;
745 notify_initialized(&self.http, server).await?;
746 post_rpc(&self.http, &server.name, http_conn, &body, CALL_TIMEOUT).await?
747 }
748 Err(error) => return Err(error),
749 }
750 }
751 };
752
753 Ok(render_tool_result(&result))
754 }
755
756 fn next_id(&self) -> i64 {
757 self.next_id.fetch_add(1, Ordering::Relaxed)
758 }
759 }
760
761 /// Flatten an MCP `tools/call` result into text. Joins text content blocks;
762 /// notes non-text blocks; honors `isError`.
763 fn render_tool_result(result: &Value) -> String {
764 let mut out = String::new();
765 if let Some(blocks) = result.get("content").and_then(Value::as_array) {
766 for block in blocks {
767 match block.get("type").and_then(Value::as_str) {
768 Some("text") => {
769 if let Some(text) = block.get("text").and_then(Value::as_str) {
770 if !out.is_empty() {
771 out.push('\n');
772 }
773 out.push_str(text);
774 }
775 }
776 Some(other) => {
777 if !out.is_empty() {
778 out.push('\n');
779 }
780 out.push_str(&format!("[{other} content omitted]"));
781 }
782 None => {}
783 }
784 }
785 }
786
787 // Some servers return only `structuredContent`; surface it if there was no
788 // textual content.
789 if out.is_empty()
790 && let Some(structured) = result.get("structuredContent")
791 {
792 out = structured.to_string();
793 }
794
795 if out.is_empty() {
796 out = "(tool returned no content)".to_string();
797 }
798
799 if result.get("isError").and_then(Value::as_bool) == Some(true) {
800 format!("Tool reported an error:\n{out}")
801 } else {
802 out
803 }
804 }
805
806 /// Truncate tool output to the context-protecting limit, with a trailing note.
807 fn truncate(text: String) -> String {
808 if text.chars().count() <= RESULT_CHAR_LIMIT {
809 return text;
810 }
811 let kept: String = text.chars().take(RESULT_CHAR_LIMIT).collect();
812 format!("{kept}\n\n[output truncated to {RESULT_CHAR_LIMIT} characters]")
813 }
814
815 // ── Transport-generic JSON-RPC dispatch ─────────────────────────────────────
816
817 /// Send a JSON-RPC request over whichever transport the server uses and return
818 /// its `result`.
819 async fn rpc_request(
820 http: &reqwest::Client,
821 conn: &ServerConn,
822 method: &str,
823 params: Value,
824 timeout: Duration,
825 ) -> Result<Value, String> {
826 match &conn.transport {
827 None => Err(format!("server '{}' is not connected", conn.name)),
828 Some(Transport::Http(http_conn)) => {
829 let body = json!({ "jsonrpc": "2.0", "id": 0, "method": method, "params": params });
830 post_rpc(http, &conn.name, http_conn, &body, timeout).await
831 }
832 Some(Transport::Stdio(stdio)) => stdio.request(method, params, timeout).await,
833 }
834 }
835
836 /// Send a JSON-RPC notification (no id, no response expected).
837 async fn rpc_notify(
838 http: &reqwest::Client,
839 conn: &ServerConn,
840 method: &str,
841 timeout: Duration,
842 ) -> Result<(), String> {
843 match &conn.transport {
844 None => Err(format!("server '{}' is not connected", conn.name)),
845 Some(Transport::Http(http_conn)) => {
846 let body = json!({ "jsonrpc": "2.0", "method": method });
847 post_notification(http, &conn.name, http_conn, &body, timeout).await
848 }
849 Some(Transport::Stdio(stdio)) => stdio.notify(method).await,
850 }
851 }
852
853 // ── stdio JSON-RPC plumbing ─────────────────────────────────────────────────
854
855 /// Spawn a stdio MCP server and start its background reader task. The child's
856 /// stderr is inherited so it lands in sigit's own log stream; the given env
857 /// vars are added on top of the inherited environment.
858 fn spawn_stdio(
859 name: &str,
860 command: &str,
861 args: &[String],
862 env: &[(String, String)],
863 ) -> Result<StdioConn, String> {
864 let mut cmd = Command::new(command);
865 cmd.args(args)
866 .stdin(std::process::Stdio::piped())
867 .stdout(std::process::Stdio::piped())
868 .stderr(std::process::Stdio::inherit())
869 .kill_on_drop(true);
870 for (key, value) in env {
871 cmd.env(key, value);
872 }
873 let mut child = cmd
874 .spawn()
875 .map_err(|error| format!("failed to spawn `{command}`: {error}"))?;
876 let stdin = child
877 .stdin
878 .take()
879 .ok_or_else(|| "child stdin was not captured".to_string())?;
880 let stdout = child
881 .stdout
882 .take()
883 .ok_or_else(|| "child stdout was not captured".to_string())?;
884
885 let shared = Arc::new(StdioShared {
886 name: name.to_string(),
887 pending: StdMutex::new(HashMap::new()),
888 dead: StdMutex::new(None),
889 child: StdMutex::new(Some(child)),
890 });
891 tokio::spawn(stdio_reader(BufReader::new(stdout), Arc::clone(&shared)));
892
893 Ok(StdioConn {
894 writer: Mutex::new(Some(stdin)),
895 shared,
896 next_id: AtomicI64::new(1),
897 })
898 }
899
900 /// Background task owning a stdio child's stdout: parses one JSON-RPC message
901 /// per line and routes each response to the pending request that carries its
902 /// id. Server-initiated requests and notifications (anything with a `method`)
903 /// are logged and ignored — siGit doesn't support server→client calls. On EOF
904 /// or a read error the connection is marked dead, which fails every in-flight
905 /// request and reaps the child.
906 async fn stdio_reader(mut stdout: BufReader<ChildStdout>, shared: Arc<StdioShared>) {
907 let mut line = String::new();
908 loop {
909 line.clear();
910 match stdout.read_line(&mut line).await {
911 Ok(0) => {
912 shared.mark_dead("server closed its stdout (process exited)");
913 return;
914 }
915 Ok(_) => {}
916 Err(error) => {
917 shared.mark_dead(&format!("read error: {error}"));
918 return;
919 }
920 }
921 let trimmed = line.trim();
922 if trimmed.is_empty() {
923 continue;
924 }
925 let message: Value = match serde_json::from_str(trimmed) {
926 Ok(message) => message,
927 Err(error) => {
928 log::warn!("mcp: '{}' sent a non-JSON line: {error}", shared.name);
929 continue;
930 }
931 };
932 if let Some(method) = message.get("method").and_then(Value::as_str) {
933 log::debug!(
934 "mcp: ignoring server-initiated '{method}' from '{}'",
935 shared.name
936 );
937 continue;
938 }
939 let Some(id) = message.get("id").and_then(Value::as_i64) else {
940 log::warn!(
941 "mcp: '{}' sent a response without a usable id; ignoring",
942 shared.name
943 );
944 continue;
945 };
946 let waiter = shared.pending.lock().unwrap().remove(&id);
947 match waiter {
948 Some(sender) => {
949 let _ = sender.send(message);
950 }
951 None => log::debug!(
952 "mcp: '{}' answered unknown/expired request id {id}; ignoring",
953 shared.name
954 ),
955 }
956 }
957 }
958
959 impl StdioConn {
960 /// Send a JSON-RPC request and await its response, correlated by id. Fails
961 /// fast (in-band, never panicking) when the child has died.
962 async fn request(
963 &self,
964 method: &str,
965 params: Value,
966 timeout: Duration,
967 ) -> Result<Value, String> {
968 let name = &self.shared.name;
969 if let Some(reason) = self.shared.dead_reason() {
970 return Err(format!("stdio server '{name}' is not running: {reason}"));
971 }
972
973 let id = self.next_id.fetch_add(1, Ordering::Relaxed);
974 let body = json!({ "jsonrpc": "2.0", "id": id, "method": method, "params": params });
975 let (sender, receiver) = oneshot::channel();
976 self.shared.pending.lock().unwrap().insert(id, sender);
977
978 if let Err(error) = self.write_line(&body).await {
979 self.shared.pending.lock().unwrap().remove(&id);
980 self.shared.mark_dead(&error);
981 return Err(format!("stdio server '{name}': {error}"));
982 }
983
984 let message = match tokio::time::timeout(timeout, receiver).await {
985 Ok(Ok(message)) => message,
986 // Our sender was dropped: the connection died mid-request.
987 Ok(Err(_)) => {
988 let reason = self
989 .shared
990 .dead_reason()
991 .unwrap_or_else(|| "connection closed".to_string());
992 return Err(format!("stdio server '{name}' is not running: {reason}"));
993 }
994 Err(_) => {
995 self.shared.pending.lock().unwrap().remove(&id);
996 return Err(format!(
997 "request to stdio server '{name}' timed out after {}s",
998 timeout.as_secs()
999 ));
1000 }
1001 };
1002
1003 if let Some(error) = message.get("error") {
1004 let code = error.get("code").and_then(Value::as_i64).unwrap_or(0);
1005 let msg = error
1006 .get("message")
1007 .and_then(Value::as_str)
1008 .unwrap_or("unknown error");
1009 return Err(format!("'{name}' JSON-RPC error {code}: {msg}"));
1010 }
1011 message
1012 .get("result")
1013 .cloned()
1014 .ok_or_else(|| format!("response from '{name}' had no result"))
1015 }
1016
1017 /// Send a JSON-RPC notification (no id, no response).
1018 async fn notify(&self, method: &str) -> Result<(), String> {
1019 let body = json!({ "jsonrpc": "2.0", "method": method });
1020 if let Err(error) = self.write_line(&body).await {
1021 self.shared.mark_dead(&error);
1022 return Err(format!("stdio server '{}': {error}", self.shared.name));
1023 }
1024 Ok(())
1025 }
1026
1027 /// Write one newline-delimited JSON-RPC message. The writer mutex keeps
1028 /// concurrent requests from interleaving bytes on the pipe.
1029 async fn write_line(&self, body: &Value) -> Result<(), String> {
1030 let mut guard = self.writer.lock().await;
1031 let Some(writer) = guard.as_mut() else {
1032 return Err("stdin already closed".to_string());
1033 };
1034 let mut line = body.to_string();
1035 line.push('\n');
1036 let result = async {
1037 writer.write_all(line.as_bytes()).await?;
1038 writer.flush().await
1039 }
1040 .await;
1041 if let Err(error) = result {
1042 // A broken pipe is unrecoverable; drop the writer so later calls
1043 // fail fast.
1044 *guard = None;
1045 return Err(format!("write failed: {error}"));
1046 }
1047 Ok(())
1048 }
1049 }
1050
1051 // ── Streamable HTTP JSON-RPC plumbing ───────────────────────────────────────
1052
1053 /// POST a JSON-RPC request and return its `result`. Handles both an
1054 /// `application/json` body and a `text/event-stream` (SSE) reply, captures the
1055 /// session id from the response headers, and maps a JSON-RPC `error` to `Err`.
1056 async fn post_rpc(
1057 http: &reqwest::Client,
1058 name: &str,
1059 conn: &HttpConn,
1060 body: &Value,
1061 timeout: Duration,
1062 ) -> Result<Value, String> {
1063 // Give every outbound request a fresh id; the on-the-wire id in `body` is a
1064 // placeholder we overwrite so callers don't have to thread a counter.
1065 let mut body = body.clone();
1066 if body.get("id").is_some()
1067 && let Some(mcp) = MCP.get()
1068 {
1069 body["id"] = json!(mcp.next_id());
1070 }
1071
1072 let response = build_request(http, conn, &body, timeout)
1073 .await
1074 .send()
1075 .await
1076 .map_err(|error| format!("request to {} failed: {error}", conn.url))?;
1077
1078 // Persist the session id the server assigns on initialize.
1079 if let Some(session) = response
1080 .headers()
1081 .get("mcp-session-id")
1082 .and_then(|value| value.to_str().ok())
1083 .map(str::to_string)
1084 {
1085 *conn.session_id.lock().await = Some(session);
1086 }
1087
1088 let status = response.status();
1089 let content_type = response
1090 .headers()
1091 .get(reqwest::header::CONTENT_TYPE)
1092 .and_then(|value| value.to_str().ok())
1093 .unwrap_or("")
1094 .to_string();
1095
1096 if !status.is_success() {
1097 let detail = response.text().await.unwrap_or_default();
1098 let detail: String = detail.chars().take(500).collect();
1099 return Err(format!(
1100 "server '{name}' returned {}: {detail}",
1101 status.as_u16()
1102 ));
1103 }
1104
1105 let text = response
1106 .text()
1107 .await
1108 .map_err(|error| format!("reading response from '{name}': {error}"))?;
1109
1110 let message = if content_type.contains("text/event-stream") {
1111 parse_sse_response(&text)
1112 .ok_or_else(|| format!("no JSON-RPC message in SSE reply from '{name}'"))?
1113 } else {
1114 serde_json::from_str::<Value>(&text)
1115 .map_err(|error| format!("parsing response from '{name}': {error}"))?
1116 };
1117
1118 if let Some(error) = message.get("error") {
1119 let code = error.get("code").and_then(Value::as_i64).unwrap_or(0);
1120 let msg = error
1121 .get("message")
1122 .and_then(Value::as_str)
1123 .unwrap_or("unknown error");
1124 return Err(format!("'{name}' JSON-RPC error {code}: {msg}"));
1125 }
1126
1127 message
1128 .get("result")
1129 .cloned()
1130 .ok_or_else(|| format!("response from '{name}' had no result"))
1131 }
1132
1133 /// POST a JSON-RPC notification (no id, no response expected). A non-success
1134 /// status is an error; an empty 202 body is the normal case.
1135 async fn post_notification(
1136 http: &reqwest::Client,
1137 name: &str,
1138 conn: &HttpConn,
1139 body: &Value,
1140 timeout: Duration,
1141 ) -> Result<(), String> {
1142 let response = build_request(http, conn, body, timeout)
1143 .await
1144 .send()
1145 .await
1146 .map_err(|error| format!("notification to {} failed: {error}", conn.url))?;
1147 if !response.status().is_success() {
1148 return Err(format!(
1149 "server '{name}' rejected notification: {}",
1150 response.status().as_u16()
1151 ));
1152 }
1153 Ok(())
1154 }
1155
1156 /// Build a request carrying the MCP headers: the dual `Accept`, the JSON body,
1157 /// the configured static headers, the negotiated protocol version, and the
1158 /// session id once we have one.
1159 async fn build_request(
1160 http: &reqwest::Client,
1161 conn: &HttpConn,
1162 body: &Value,
1163 timeout: Duration,
1164 ) -> reqwest::RequestBuilder {
1165 let mut request = http
1166 .post(&conn.url)
1167 .timeout(timeout)
1168 .header(
1169 reqwest::header::ACCEPT,
1170 "application/json, text/event-stream",
1171 )
1172 .header("MCP-Protocol-Version", PROTOCOL_VERSION)
1173 .json(body);
1174
1175 for (key, value) in &conn.headers {
1176 request = request.header(key.as_str(), value.as_str());
1177 }
1178 if let Some(session) = conn.session_id.lock().await.as_ref() {
1179 request = request.header("Mcp-Session-Id", session.as_str());
1180 }
1181 request
1182 }
1183
1184 /// Extract the first JSON-RPC message from an SSE body. SSE frames are separated
1185 /// by blank lines; each `data:` line contributes to the frame's payload. For a
1186 /// single request/response exchange the server sends one `message` event whose
1187 /// data is the JSON-RPC response.
1188 fn parse_sse_response(body: &str) -> Option<Value> {
1189 let mut data = String::new();
1190 for line in body.lines() {
1191 if let Some(rest) = line.strip_prefix("data:") {
1192 if !data.is_empty() {
1193 data.push('\n');
1194 }
1195 data.push_str(rest.strip_prefix(' ').unwrap_or(rest));
1196 } else if line.trim().is_empty() && !data.is_empty() {
1197 // End of an event — try to parse it as a JSON-RPC message.
1198 if let Ok(value) = serde_json::from_str::<Value>(&data)
1199 && (value.get("result").is_some() || value.get("error").is_some())
1200 {
1201 return Some(value);
1202 }
1203 data.clear();
1204 }
1205 }
1206 // Trailing event without a closing blank line.
1207 if !data.is_empty()
1208 && let Ok(value) = serde_json::from_str::<Value>(&data)
1209 && (value.get("result").is_some() || value.get("error").is_some())
1210 {
1211 return Some(value);
1212 }
1213 None
1214 }
1215
1216 // ── Status reporting (`/mcp`) ────────────────────────────────────────────────
1217
1218 /// Human-readable summary of configured MCP servers and their tools, for the
1219 /// `/mcp` slash command. Shows the URL for HTTP servers and the command line
1220 /// for stdio servers.
1221 pub fn status_summary() -> String {
1222 let Some(mcp) = MCP.get() else {
1223 return "MCP is not initialized.".to_string();
1224 };
1225 if mcp.servers.is_empty() {
1226 return "No MCP servers configured. Add one in ~/.config/sigit/mcp.toml \
1227 or .sigit/mcp.toml. See https://modelcontextprotocol.io."
1228 .to_string();
1229 }
1230
1231 let total_tools: usize = mcp.servers.iter().map(|s| s.tools.len()).sum();
1232 let mut lines = vec![format!(
1233 "{} MCP server(s), {total_tools} tool(s) available:",
1234 mcp.servers.len()
1235 )];
1236 for server in &mcp.servers {
1237 match &server.error {
1238 Some(error) => lines.push(format!(
1239 "- {} ({}) — unavailable: {error}",
1240 server.name, server.endpoint
1241 )),
1242 None => {
1243 lines.push(format!(
1244 "- {} ({}) — {} tool(s)",
1245 server.name,
1246 server.endpoint,
1247 server.tools.len()
1248 ));
1249 for tool in &server.tools {
1250 lines.push(format!(" • {}", tool.full_name));
1251 }
1252 }
1253 }
1254 }
1255 lines.join("\n")
1256 }
1257
1258 #[cfg(test)]
1259 mod tests {
1260 use super::*;
1261
1262 #[test]
1263 fn is_mcp_tool_detects_prefix() {
1264 assert!(is_mcp_tool("mcp__sigit__search"));
1265 assert!(!is_mcp_tool("read_file"));
1266 assert!(!is_mcp_tool("skill"));
1267 }
1268
1269 #[test]
1270 fn sanitize_collapses_invalid_chars() {
1271 assert_eq!(sanitize("github"), "github");
1272 assert_eq!(sanitize("my server"), "my_server");
1273 assert_eq!(sanitize("a.b/c:d"), "a_b_c_d");
1274 assert_eq!(sanitize("keep-_ok9"), "keep-_ok9");
1275 }
1276
1277 #[test]
1278 fn parses_mcp_file_with_servers() {
1279 let toml = r#"
1280 official = false
1281
1282 [[server]]
1283 name = "github"
1284 url = "https://api.example.com/mcp"
1285
1286 [[server]]
1287 name = "disabled-one"
1288 url = "https://nope.example.com/mcp"
1289 enabled = false
1290
1291 [server.headers]
1292 Authorization = "Bearer xyz"
1293 "#;
1294 let parsed: McpFile = toml::from_str(toml).unwrap();
1295 assert_eq!(parsed.official, Some(false));
1296 assert_eq!(parsed.server.len(), 2);
1297 assert_eq!(parsed.server[0].name, "github");
1298 assert_eq!(parsed.server[1].enabled, Some(false));
1299 assert_eq!(
1300 parsed.server[1]
1301 .headers
1302 .get("Authorization")
1303 .map(String::as_str),
1304 Some("Bearer xyz")
1305 );
1306 }
1307
1308 #[test]
1309 fn parses_stdio_server_with_args_and_env() {
1310 let toml = r#"
1311 [[server]]
1312 name = "fs"
1313 command = "npx"
1314 args = ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"]
1315
1316 [server.env]
1317 LOG_LEVEL = "debug"
1318 TOKEN = "abc"
1319 "#;
1320 let parsed: McpFile = toml::from_str(toml).unwrap();
1321 assert_eq!(parsed.server.len(), 1);
1322 let entry = &parsed.server[0];
1323 assert_eq!(entry.command.as_deref(), Some("npx"));
1324 assert_eq!(entry.args.len(), 3);
1325 assert_eq!(
1326 entry.env.get("LOG_LEVEL").map(String::as_str),
1327 Some("debug")
1328 );
1329 assert_eq!(entry.env.get("TOKEN").map(String::as_str), Some("abc"));
1330
1331 let def = entry.transport_def().expect("valid stdio entry");
1332 match def {
1333 TransportDef::Stdio { command, args, env } => {
1334 assert_eq!(command, "npx");
1335 assert_eq!(args[0], "-y");
1336 assert_eq!(env.len(), 2);
1337 }
1338 TransportDef::Http { .. } => panic!("expected a stdio transport"),
1339 }
1340 }
1341
1342 #[test]
1343 fn entry_with_url_and_command_is_a_config_error() {
1344 let toml = r#"
1345 [[server]]
1346 name = "confused"
1347 url = "https://example.com/mcp"
1348 command = "npx"
1349 "#;
1350 let parsed: McpFile = toml::from_str(toml).unwrap();
1351 let error = parsed.server[0].transport_def().unwrap_err();
1352 assert!(error.contains("both"), "unexpected error: {error}");
1353 }
1354
1355 #[test]
1356 fn entry_with_neither_url_nor_command_is_a_config_error() {
1357 let toml = r#"
1358 [[server]]
1359 name = "empty"
1360 "#;
1361 let parsed: McpFile = toml::from_str(toml).unwrap();
1362 let error = parsed.server[0].transport_def().unwrap_err();
1363 assert!(error.contains("needs"), "unexpected error: {error}");
1364 }
1365
1366 #[test]
1367 fn blank_url_or_command_counts_as_absent() {
1368 let toml = r#"
1369 [[server]]
1370 name = "blank"
1371 url = " "
1372 command = "server-bin"
1373 "#;
1374 let parsed: McpFile = toml::from_str(toml).unwrap();
1375 // A blank url is treated as absent, so this resolves to stdio.
1376 match parsed.server[0].transport_def().expect("stdio") {
1377 TransportDef::Stdio { command, .. } => assert_eq!(command, "server-bin"),
1378 TransportDef::Http { .. } => panic!("expected stdio"),
1379 }
1380 }
1381
1382 #[test]
1383 fn endpoint_renders_url_or_command_line() {
1384 let http = TransportDef::Http {
1385 url: "https://example.com/mcp".into(),
1386 headers: vec![],
1387 };
1388 assert_eq!(http.endpoint(), "https://example.com/mcp");
1389
1390 let stdio = TransportDef::Stdio {
1391 command: "npx".into(),
1392 args: vec!["-y".into(), "server-fs".into()],
1393 env: vec![],
1394 };
1395 assert_eq!(stdio.endpoint(), "npx -y server-fs");
1396 }
1397
1398 #[test]
1399 fn upsert_replaces_same_name() {
1400 let mut defs = vec![ServerDef {
1401 name: "a".into(),
1402 transport: TransportDef::Http {
1403 url: "u1".into(),
1404 headers: vec![],
1405 },
1406 }];
1407 upsert(
1408 &mut defs,
1409 ServerDef {
1410 name: "a".into(),
1411 transport: TransportDef::Http {
1412 url: "u2".into(),
1413 headers: vec![],
1414 },
1415 },
1416 );
1417 assert_eq!(defs.len(), 1);
1418 assert_eq!(defs[0].transport.endpoint(), "u2");
1419 }
1420
1421 #[test]
1422 fn parse_sse_extracts_jsonrpc_response() {
1423 let body =
1424 "event: message\ndata: {\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"ok\":true}}\n\n";
1425 let value = parse_sse_response(body).expect("a message");
1426 assert_eq!(value["result"]["ok"], json!(true));
1427 }
1428
1429 #[test]
1430 fn parse_sse_handles_no_trailing_blank_line() {
1431 let body = "data: {\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{}}";
1432 assert!(parse_sse_response(body).is_some());
1433 }
1434
1435 #[test]
1436 fn parse_sse_ignores_non_response_frames() {
1437 // A lone notification (no result/error) shouldn't be mistaken for the response.
1438 let body = "data: {\"jsonrpc\":\"2.0\",\"method\":\"ping\"}\n\n";
1439 assert!(parse_sse_response(body).is_none());
1440 }
1441
1442 #[test]
1443 fn render_result_joins_text_blocks() {
1444 let result = json!({
1445 "content": [
1446 { "type": "text", "text": "line one" },
1447 { "type": "text", "text": "line two" }
1448 ]
1449 });
1450 assert_eq!(render_tool_result(&result), "line one\nline two");
1451 }
1452
1453 #[test]
1454 fn render_result_marks_errors_and_non_text() {
1455 let result = json!({
1456 "isError": true,
1457 "content": [
1458 { "type": "text", "text": "boom" },
1459 { "type": "image", "data": "..." }
1460 ]
1461 });
1462 let rendered = render_tool_result(&result);
1463 assert!(rendered.starts_with("Tool reported an error:"));
1464 assert!(rendered.contains("boom"));
1465 assert!(rendered.contains("[image content omitted]"));
1466 }
1467
1468 #[test]
1469 fn render_result_falls_back_to_structured_content() {
1470 let result = json!({ "structuredContent": { "value": 42 } });
1471 assert!(render_tool_result(&result).contains("42"));
1472 }
1473
1474 #[test]
1475 fn truncate_caps_long_output() {
1476 let long = "x".repeat(RESULT_CHAR_LIMIT + 100);
1477 let out = truncate(long);
1478 assert!(out.contains("[output truncated"));
1479 }
1480
1481 #[test]
1482 fn tool_specs_empty_before_init() {
1483 // Without init() the global is unset; this must not panic.
1484 assert!(super::tool_specs().is_empty() || MCP.get().is_some());
1485 }
1486 }