@setoelkahfi / sigit
Code Commits Pulls CI/CD 232
main
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 954 lines 32.5 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 //! Transport: the modern **Streamable HTTP** transport — a single HTTP endpoint
10 //! the client POSTs JSON-RPC 2.0 messages to. The server answers either with a
11 //! single `application/json` body or a `text/event-stream` (SSE) stream that
12 //! carries the JSON-RPC response. Both are handled here. stdio transport is not
13 //! supported (siGit Code never spawns child processes for inference).
14 //!
15 //! ## The official server
16 //!
17 //! siGit Code bakes in its official MCP server at `<cloud>/mcp` (default
18 //! `https://sigit.si/api/v1/mcp`, following `SIGIT_CLOUD_URL`). When the user is
19 //! signed in (`sigit login`) the cloud session token is sent as a bearer
20 //! credential. Additional servers are configured in `mcp.toml` (see
21 //! [`load_configs`]).
22 //!
23 //! ## Lifecycle
24 //!
25 //! Discovery is best-effort and happens once at startup via [`init`]: each
26 //! configured server is contacted concurrently (with a per-server timeout),
27 //! runs the `initialize` handshake, and has its `tools/list` cached. A server
28 //! that fails to connect is recorded with its error and simply contributes no
29 //! tools — it never blocks startup or the rest of the agent. The result is
30 //! stored in a process-global so the synchronous tool-spec builders
31 //! ([`tool_specs`]) and the async dispatch ([`call_tool`]) can both read it.
32 //!
33 //! Tools are namespaced `mcp__<server>__<tool>` so they never collide with
34 //! built-in tools or with each other across servers. This mirrors the
35 //! convention used by other MCP-aware agents.
36 //!
37 //! Like the rest of the backend seam, MCP is wired up only through the
38 //! interactive client and the ACP agent loop. On non-Unix targets a few helpers
39 //! are unused, so the dead-code lint is suppressed there only.
40 #![cfg_attr(not(unix), allow(dead_code))]
41
42 use std::collections::BTreeMap;
43 use std::path::PathBuf;
44 use std::sync::OnceLock;
45 use std::sync::atomic::{AtomicI64, Ordering};
46 use std::time::Duration;
47
48 use serde::Deserialize;
49 use serde_json::{Value, json};
50 use tokio::sync::Mutex;
51
52 use crate::backend::ToolSpec;
53
54 /// Prefix marking a tool as MCP-provided. The full name is
55 /// `mcp__<server>__<tool>`.
56 pub const MCP_PREFIX: &str = "mcp__";
57
58 /// JSON-RPC / MCP protocol version we advertise in the handshake.
59 const PROTOCOL_VERSION: &str = "2025-06-18";
60
61 /// Per-server budget for the connect + `initialize` + `tools/list` handshake at
62 /// startup. Bounds how long an unreachable server can delay startup; servers are
63 /// contacted concurrently, so this is the worst case for the whole set, not the
64 /// sum.
65 const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(8);
66
67 /// Overall request timeout for an individual `tools/call`. Generous, since an
68 /// MCP tool may do real work server-side.
69 const CALL_TIMEOUT: Duration = Duration::from_secs(120);
70
71 /// Cap on the characters returned from a single tool call, so a chatty server
72 /// can't blow up the model's context. Matches the spirit of the file-read cap.
73 const RESULT_CHAR_LIMIT: usize = 30_000;
74
75 // ── Public types ──────────────────────────────────────────────────────────────
76
77 /// A tool discovered on an MCP server, in siGit's flattened form.
78 #[derive(Debug, Clone)]
79 struct McpTool {
80 /// Namespaced name exposed to the model: `mcp__<server>__<tool>`.
81 full_name: String,
82 /// The tool's name as the server knows it (sent back in `tools/call`).
83 remote_name: String,
84 /// Human/model-facing description, prefixed with the server name.
85 description: String,
86 /// JSON Schema for the tool's arguments, encoded as a string.
87 parameters_schema: String,
88 }
89
90 /// A configured MCP server and its live connection state.
91 struct ServerConn {
92 /// Sanitized server name used in tool namespacing and the `/mcp` listing.
93 name: String,
94 /// Streamable HTTP endpoint (the single POST URL).
95 url: String,
96 /// Extra headers sent on every request (e.g. `Authorization`).
97 headers: Vec<(String, String)>,
98 /// Session id handed back by the server on `initialize`, echoed on every
99 /// later request via the `Mcp-Session-Id` header.
100 session_id: Mutex<Option<String>>,
101 /// Tools discovered at startup. Empty when the server failed to connect.
102 tools: Vec<McpTool>,
103 /// Connection error, if the handshake failed. Surfaced by `/mcp`.
104 error: Option<String>,
105 }
106
107 /// The process-global MCP state: a shared HTTP client plus every configured
108 /// server.
109 struct Mcp {
110 http: reqwest::Client,
111 servers: Vec<ServerConn>,
112 next_id: AtomicI64,
113 }
114
115 static MCP: OnceLock<Mcp> = OnceLock::new();
116
117 // ── Configuration ───────────────────────────────────────────────────────────
118
119 /// Default endpoint of the official siGit Code MCP server, derived from the
120 /// cloud base URL so `SIGIT_CLOUD_URL` (dev) carries over.
121 fn official_url() -> String {
122 format!(
123 "{}/mcp",
124 crate::provider::cloud_base_url().trim_end_matches('/')
125 )
126 }
127
128 /// A server entry as written in `mcp.toml`.
129 #[derive(Debug, Deserialize)]
130 struct ServerEntry {
131 name: String,
132 url: String,
133 /// Set `enabled = false` to keep an entry in the file but skip connecting.
134 #[serde(default)]
135 enabled: Option<bool>,
136 /// Static headers, e.g. `Authorization = "Bearer ..."`.
137 #[serde(default)]
138 headers: BTreeMap<String, String>,
139 }
140
141 /// The `mcp.toml` schema.
142 #[derive(Debug, Default, Deserialize)]
143 struct McpFile {
144 /// Include the baked-in official server. Defaults to `true`; set `false` to
145 /// opt out.
146 #[serde(default)]
147 official: Option<bool>,
148 #[serde(default)]
149 server: Vec<ServerEntry>,
150 }
151
152 /// A resolved server definition, before connecting.
153 #[derive(Debug, Clone)]
154 struct ServerDef {
155 name: String,
156 url: String,
157 headers: Vec<(String, String)>,
158 }
159
160 /// Config files to read, in priority order (later wins on a name clash):
161 /// global `$SIGIT_CONFIG_DIR/mcp.toml`, then project-local `<cwd>/.sigit/mcp.toml`.
162 fn config_paths() -> Vec<PathBuf> {
163 let mut paths = Vec::new();
164 if let Some(dir) = sigit_config_dir() {
165 paths.push(dir.join("mcp.toml"));
166 }
167 if let Ok(cwd) = std::env::current_dir() {
168 paths.push(cwd.join(".sigit").join("mcp.toml"));
169 }
170 paths
171 }
172
173 fn sigit_config_dir() -> Option<PathBuf> {
174 if let Ok(dir) = std::env::var("SIGIT_CONFIG_DIR")
175 && !dir.is_empty()
176 {
177 return Some(PathBuf::from(dir));
178 }
179 std::env::var("HOME")
180 .ok()
181 .map(|home| PathBuf::from(home).join(".config").join("sigit"))
182 }
183
184 /// Resolve the full set of servers to connect to: the baked-in official server
185 /// (unless opted out) plus any from `mcp.toml`. Project-local entries override
186 /// global ones, and a user entry named `sigit` overrides the official default.
187 fn load_configs() -> Vec<ServerDef> {
188 // Global escape hatch: `SIGIT_MCP=off` disables MCP entirely.
189 if let Ok(value) = std::env::var("SIGIT_MCP")
190 && matches!(
191 value.trim().to_ascii_lowercase().as_str(),
192 "off" | "0" | "false" | "no" | "disabled"
193 )
194 {
195 log::info!("mcp: disabled via SIGIT_MCP");
196 return Vec::new();
197 }
198
199 let mut include_official = true;
200 // De-duplicated by sanitized name; a later config file overrides an earlier
201 // one for the same name (project-local wins over global).
202 let mut defs: Vec<ServerDef> = Vec::new();
203
204 for path in config_paths() {
205 let Ok(contents) = std::fs::read_to_string(&path) else {
206 continue;
207 };
208 let parsed: McpFile = match toml::from_str(&contents) {
209 Ok(parsed) => parsed,
210 Err(error) => {
211 log::warn!("mcp: ignoring {}: {error}", path.display());
212 continue;
213 }
214 };
215 if let Some(official) = parsed.official {
216 include_official = official;
217 }
218 for entry in parsed.server {
219 if entry.enabled == Some(false) {
220 continue;
221 }
222 let name = sanitize(&entry.name);
223 if name.is_empty() || entry.url.trim().is_empty() {
224 log::warn!(
225 "mcp: skipping server with empty name/url in {}",
226 path.display()
227 );
228 continue;
229 }
230 let headers = entry.headers.into_iter().collect();
231 upsert(
232 &mut defs,
233 ServerDef {
234 name,
235 url: entry.url.trim().to_string(),
236 headers,
237 },
238 );
239 }
240 }
241
242 // The official server can also be disabled with SIGIT_MCP_OFFICIAL=off.
243 if let Ok(value) = std::env::var("SIGIT_MCP_OFFICIAL")
244 && matches!(
245 value.trim().to_ascii_lowercase().as_str(),
246 "off" | "0" | "false" | "no"
247 )
248 {
249 include_official = false;
250 }
251
252 // Add the baked-in official server, but never clobber a user-defined entry
253 // named `sigit` — an explicit config (e.g. a custom URL or headers) wins.
254 if include_official && !defs.iter().any(|d| d.name == "sigit") {
255 let mut headers = Vec::new();
256 if let Some(token) = crate::credentials::load_token() {
257 headers.push(("Authorization".to_string(), format!("Bearer {token}")));
258 }
259 defs.push(ServerDef {
260 name: "sigit".to_string(),
261 url: official_url(),
262 headers,
263 });
264 }
265
266 defs
267 }
268
269 /// Insert `def`, replacing any existing entry with the same name.
270 fn upsert(defs: &mut Vec<ServerDef>, def: ServerDef) {
271 if let Some(slot) = defs.iter_mut().find(|d| d.name == def.name) {
272 *slot = def;
273 } else {
274 defs.push(def);
275 }
276 }
277
278 /// Sanitize a name into the `[a-zA-Z0-9_-]` set tool names are restricted to,
279 /// collapsing anything else to `_`.
280 fn sanitize(raw: &str) -> String {
281 raw.trim()
282 .chars()
283 .map(|c| {
284 if c.is_ascii_alphanumeric() || c == '_' || c == '-' {
285 c
286 } else {
287 '_'
288 }
289 })
290 .collect()
291 }
292
293 // ── Startup / discovery ─────────────────────────────────────────────────────
294
295 /// Connect to every configured server and cache the tools they expose. Idempotent
296 /// and best-effort: a server that can't be reached is recorded with its error and
297 /// contributes no tools. Safe to call from either entry point; only the first
298 /// call does work.
299 pub async fn init() {
300 if MCP.get().is_some() {
301 return;
302 }
303
304 let defs = load_configs();
305 let http = reqwest::Client::builder()
306 .timeout(CALL_TIMEOUT)
307 .user_agent(concat!(
308 "sigit/",
309 env!("CARGO_PKG_VERSION"),
310 " (mcp-client)"
311 ))
312 .build()
313 .unwrap_or_default();
314
315 // Contact servers concurrently so one slow/unreachable host doesn't serialize
316 // the rest. Each handshake is bounded by HANDSHAKE_TIMEOUT.
317 let connects = defs.into_iter().map(|def| {
318 let http = http.clone();
319 async move { connect(&http, def).await }
320 });
321 let servers = futures::future::join_all(connects).await;
322
323 for server in &servers {
324 match &server.error {
325 Some(error) => log::warn!("mcp: server '{}' unavailable: {error}", server.name),
326 None => log::info!(
327 "mcp: server '{}' ready, {} tool(s)",
328 server.name,
329 server.tools.len()
330 ),
331 }
332 }
333
334 let _ = MCP.set(Mcp {
335 http,
336 servers,
337 next_id: AtomicI64::new(1),
338 });
339 }
340
341 /// Run the handshake against one server and collect its tools. Always returns a
342 /// `ServerConn`; failures land in its `error` field rather than propagating.
343 async fn connect(http: &reqwest::Client, def: ServerDef) -> ServerConn {
344 let mut conn = ServerConn {
345 name: def.name.clone(),
346 url: def.url.clone(),
347 headers: def.headers.clone(),
348 session_id: Mutex::new(None),
349 tools: Vec::new(),
350 error: None,
351 };
352
353 let handshake = tokio::time::timeout(HANDSHAKE_TIMEOUT, async {
354 // initialize → notifications/initialized → tools/list
355 initialize(http, &conn).await?;
356 notify_initialized(http, &conn).await?;
357 list_tools(http, &conn).await
358 })
359 .await;
360
361 match handshake {
362 Ok(Ok(tools)) => conn.tools = tools,
363 Ok(Err(error)) => conn.error = Some(error),
364 Err(_) => conn.error = Some(format!("timed out after {}s", HANDSHAKE_TIMEOUT.as_secs())),
365 }
366
367 conn
368 }
369
370 /// The `initialize` request: negotiate protocol version and capture the session
371 /// id from the response headers (handled inside [`post_rpc`]).
372 async fn initialize(http: &reqwest::Client, conn: &ServerConn) -> Result<(), String> {
373 let body = json!({
374 "jsonrpc": "2.0",
375 "id": 0,
376 "method": "initialize",
377 "params": {
378 "protocolVersion": PROTOCOL_VERSION,
379 "capabilities": {},
380 "clientInfo": { "name": "sigit", "version": env!("CARGO_PKG_VERSION") }
381 }
382 });
383 post_rpc(http, conn, &body, HANDSHAKE_TIMEOUT).await?;
384 Ok(())
385 }
386
387 /// The `notifications/initialized` notification. Servers expect it before
388 /// fielding requests; it carries no id and yields a 202 with no body.
389 async fn notify_initialized(http: &reqwest::Client, conn: &ServerConn) -> Result<(), String> {
390 let body = json!({ "jsonrpc": "2.0", "method": "notifications/initialized" });
391 post_notification(http, conn, &body, HANDSHAKE_TIMEOUT).await
392 }
393
394 /// `tools/list`, following `nextCursor` pagination, mapped into [`McpTool`]s.
395 async fn list_tools(http: &reqwest::Client, conn: &ServerConn) -> Result<Vec<McpTool>, String> {
396 let mut tools = Vec::new();
397 let mut cursor: Option<String> = None;
398
399 loop {
400 let params = match &cursor {
401 Some(c) => json!({ "cursor": c }),
402 None => json!({}),
403 };
404 let body = json!({ "jsonrpc": "2.0", "id": 0, "method": "tools/list", "params": params });
405 let result = post_rpc(http, conn, &body, HANDSHAKE_TIMEOUT).await?;
406
407 for tool in result
408 .get("tools")
409 .and_then(Value::as_array)
410 .into_iter()
411 .flatten()
412 {
413 let Some(remote_name) = tool.get("name").and_then(Value::as_str) else {
414 continue;
415 };
416 let full_name = format!("{MCP_PREFIX}{}__{}", conn.name, sanitize(remote_name));
417 if full_name.chars().count() > 64 {
418 log::warn!(
419 "mcp: tool name '{full_name}' exceeds 64 chars; some backends may reject it"
420 );
421 }
422 let remote_desc = tool
423 .get("description")
424 .and_then(Value::as_str)
425 .unwrap_or("")
426 .trim();
427 let description = if remote_desc.is_empty() {
428 format!("[MCP server '{}'] {remote_name}", conn.name)
429 } else {
430 format!("[MCP server '{}'] {remote_desc}", conn.name)
431 };
432 // `inputSchema` is a JSON Schema object; default to a permissive
433 // object schema when a server omits it.
434 let parameters_schema = tool
435 .get("inputSchema")
436 .filter(|schema| schema.is_object())
437 .cloned()
438 .unwrap_or_else(|| json!({ "type": "object" }))
439 .to_string();
440
441 tools.push(McpTool {
442 full_name,
443 remote_name: remote_name.to_string(),
444 description,
445 parameters_schema,
446 });
447 }
448
449 cursor = result
450 .get("nextCursor")
451 .and_then(Value::as_str)
452 .map(str::to_string);
453 if cursor.is_none() {
454 break;
455 }
456 }
457
458 Ok(tools)
459 }
460
461 // ── Tool exposure + dispatch ────────────────────────────────────────────────
462
463 /// Whether a tool name belongs to MCP. The dispatch in `tools::execute_tool`
464 /// uses this to route a call here.
465 pub fn is_mcp_tool(name: &str) -> bool {
466 name.starts_with(MCP_PREFIX)
467 }
468
469 /// All discovered MCP tools as agent [`ToolSpec`]s, ready to append to the
470 /// built-in tool list. Empty when MCP is uninitialized or no server exposed any.
471 pub fn tool_specs() -> Vec<ToolSpec> {
472 let Some(mcp) = MCP.get() else {
473 return Vec::new();
474 };
475 let mut specs = Vec::new();
476 for server in &mcp.servers {
477 for tool in &server.tools {
478 specs.push(ToolSpec {
479 name: tool.full_name.clone(),
480 description: tool.description.clone(),
481 parameters_schema: tool.parameters_schema.clone(),
482 });
483 }
484 }
485 specs
486 }
487
488 /// Execute an MCP tool call by name, returning text to feed back to the model.
489 /// Errors are returned as plain strings (never panics) so a failing tool degrades
490 /// to a message the model can react to, exactly like the built-in tools.
491 pub async fn call_tool(full_name: &str, arguments: &str) -> String {
492 let Some(mcp) = MCP.get() else {
493 return "Error: MCP is not initialized.".to_string();
494 };
495
496 let Some((server, tool)) = mcp.servers.iter().find_map(|s| {
497 s.tools
498 .iter()
499 .find(|t| t.full_name == full_name)
500 .map(|t| (s, t))
501 }) else {
502 return format!("Error: unknown MCP tool \"{full_name}\".");
503 };
504
505 // Arguments arrive as a JSON-encoded string; an empty/blank string means no
506 // arguments. Anything that isn't a JSON object is a model mistake.
507 let args: Value = if arguments.trim().is_empty() {
508 json!({})
509 } else {
510 match serde_json::from_str(arguments) {
511 Ok(value @ Value::Object(_)) => value,
512 Ok(_) => return "Error: tool arguments must be a JSON object.".to_string(),
513 Err(error) => return format!("Error: failed to parse arguments: {error}"),
514 }
515 };
516
517 match mcp.call(server, &tool.remote_name, args).await {
518 Ok(text) => truncate(text),
519 Err(error) => format!("Error: {error}"),
520 }
521 }
522
523 impl Mcp {
524 /// Send a `tools/call` and render the result into text. Retries once after a
525 /// re-`initialize` if the session was dropped (HTTP 404), which is how
526 /// Streamable HTTP signals an expired session.
527 async fn call(
528 &self,
529 server: &ServerConn,
530 remote_name: &str,
531 args: Value,
532 ) -> Result<String, String> {
533 let body = json!({
534 "jsonrpc": "2.0",
535 "id": 0,
536 "method": "tools/call",
537 "params": { "name": remote_name, "arguments": args }
538 });
539
540 let result = match post_rpc(&self.http, server, &body, CALL_TIMEOUT).await {
541 Ok(result) => result,
542 Err(error) if error.contains("returned 404") => {
543 // Session expired — drop it, re-handshake, and retry once.
544 *server.session_id.lock().await = None;
545 initialize(&self.http, server).await?;
546 notify_initialized(&self.http, server).await?;
547 post_rpc(&self.http, server, &body, CALL_TIMEOUT).await?
548 }
549 Err(error) => return Err(error),
550 };
551
552 Ok(render_tool_result(&result))
553 }
554
555 fn next_id(&self) -> i64 {
556 self.next_id.fetch_add(1, Ordering::Relaxed)
557 }
558 }
559
560 /// Flatten an MCP `tools/call` result into text. Joins text content blocks;
561 /// notes non-text blocks; honors `isError`.
562 fn render_tool_result(result: &Value) -> String {
563 let mut out = String::new();
564 if let Some(blocks) = result.get("content").and_then(Value::as_array) {
565 for block in blocks {
566 match block.get("type").and_then(Value::as_str) {
567 Some("text") => {
568 if let Some(text) = block.get("text").and_then(Value::as_str) {
569 if !out.is_empty() {
570 out.push('\n');
571 }
572 out.push_str(text);
573 }
574 }
575 Some(other) => {
576 if !out.is_empty() {
577 out.push('\n');
578 }
579 out.push_str(&format!("[{other} content omitted]"));
580 }
581 None => {}
582 }
583 }
584 }
585
586 // Some servers return only `structuredContent`; surface it if there was no
587 // textual content.
588 if out.is_empty()
589 && let Some(structured) = result.get("structuredContent")
590 {
591 out = structured.to_string();
592 }
593
594 if out.is_empty() {
595 out = "(tool returned no content)".to_string();
596 }
597
598 if result.get("isError").and_then(Value::as_bool) == Some(true) {
599 format!("Tool reported an error:\n{out}")
600 } else {
601 out
602 }
603 }
604
605 /// Truncate tool output to the context-protecting limit, with a trailing note.
606 fn truncate(text: String) -> String {
607 if text.chars().count() <= RESULT_CHAR_LIMIT {
608 return text;
609 }
610 let kept: String = text.chars().take(RESULT_CHAR_LIMIT).collect();
611 format!("{kept}\n\n[output truncated to {RESULT_CHAR_LIMIT} characters]")
612 }
613
614 // ── Streamable HTTP JSON-RPC plumbing ───────────────────────────────────────
615
616 /// POST a JSON-RPC request and return its `result`. Handles both an
617 /// `application/json` body and a `text/event-stream` (SSE) reply, captures the
618 /// session id from the response headers, and maps a JSON-RPC `error` to `Err`.
619 async fn post_rpc(
620 http: &reqwest::Client,
621 conn: &ServerConn,
622 body: &Value,
623 timeout: Duration,
624 ) -> Result<Value, String> {
625 // Give every outbound request a fresh id; the on-the-wire id in `body` is a
626 // placeholder we overwrite so callers don't have to thread a counter.
627 let mut body = body.clone();
628 if body.get("id").is_some()
629 && let Some(mcp) = MCP.get()
630 {
631 body["id"] = json!(mcp.next_id());
632 }
633
634 let response = build_request(http, conn, &body, timeout)
635 .await
636 .send()
637 .await
638 .map_err(|error| format!("request to {} failed: {error}", conn.url))?;
639
640 // Persist the session id the server assigns on initialize.
641 if let Some(session) = response
642 .headers()
643 .get("mcp-session-id")
644 .and_then(|value| value.to_str().ok())
645 .map(str::to_string)
646 {
647 *conn.session_id.lock().await = Some(session);
648 }
649
650 let status = response.status();
651 let content_type = response
652 .headers()
653 .get(reqwest::header::CONTENT_TYPE)
654 .and_then(|value| value.to_str().ok())
655 .unwrap_or("")
656 .to_string();
657
658 if !status.is_success() {
659 let detail = response.text().await.unwrap_or_default();
660 let detail: String = detail.chars().take(500).collect();
661 return Err(format!(
662 "server '{}' returned {}: {detail}",
663 conn.name,
664 status.as_u16()
665 ));
666 }
667
668 let text = response
669 .text()
670 .await
671 .map_err(|error| format!("reading response from '{}': {error}", conn.name))?;
672
673 let message = if content_type.contains("text/event-stream") {
674 parse_sse_response(&text)
675 .ok_or_else(|| format!("no JSON-RPC message in SSE reply from '{}'", conn.name))?
676 } else {
677 serde_json::from_str::<Value>(&text)
678 .map_err(|error| format!("parsing response from '{}': {error}", conn.name))?
679 };
680
681 if let Some(error) = message.get("error") {
682 let code = error.get("code").and_then(Value::as_i64).unwrap_or(0);
683 let msg = error
684 .get("message")
685 .and_then(Value::as_str)
686 .unwrap_or("unknown error");
687 return Err(format!("'{}' JSON-RPC error {code}: {msg}", conn.name));
688 }
689
690 message
691 .get("result")
692 .cloned()
693 .ok_or_else(|| format!("response from '{}' had no result", conn.name))
694 }
695
696 /// POST a JSON-RPC notification (no id, no response expected). A non-success
697 /// status is an error; an empty 202 body is the normal case.
698 async fn post_notification(
699 http: &reqwest::Client,
700 conn: &ServerConn,
701 body: &Value,
702 timeout: Duration,
703 ) -> Result<(), String> {
704 let response = build_request(http, conn, body, timeout)
705 .await
706 .send()
707 .await
708 .map_err(|error| format!("notification to {} failed: {error}", conn.url))?;
709 if !response.status().is_success() {
710 return Err(format!(
711 "server '{}' rejected notification: {}",
712 conn.name,
713 response.status().as_u16()
714 ));
715 }
716 Ok(())
717 }
718
719 /// Build a request carrying the MCP headers: the dual `Accept`, the JSON body,
720 /// the configured static headers, the negotiated protocol version, and the
721 /// session id once we have one.
722 async fn build_request(
723 http: &reqwest::Client,
724 conn: &ServerConn,
725 body: &Value,
726 timeout: Duration,
727 ) -> reqwest::RequestBuilder {
728 let mut request = http
729 .post(&conn.url)
730 .timeout(timeout)
731 .header(
732 reqwest::header::ACCEPT,
733 "application/json, text/event-stream",
734 )
735 .header("MCP-Protocol-Version", PROTOCOL_VERSION)
736 .json(body);
737
738 for (key, value) in &conn.headers {
739 request = request.header(key.as_str(), value.as_str());
740 }
741 if let Some(session) = conn.session_id.lock().await.as_ref() {
742 request = request.header("Mcp-Session-Id", session.as_str());
743 }
744 request
745 }
746
747 /// Extract the first JSON-RPC message from an SSE body. SSE frames are separated
748 /// by blank lines; each `data:` line contributes to the frame's payload. For a
749 /// single request/response exchange the server sends one `message` event whose
750 /// data is the JSON-RPC response.
751 fn parse_sse_response(body: &str) -> Option<Value> {
752 let mut data = String::new();
753 for line in body.lines() {
754 if let Some(rest) = line.strip_prefix("data:") {
755 if !data.is_empty() {
756 data.push('\n');
757 }
758 data.push_str(rest.strip_prefix(' ').unwrap_or(rest));
759 } else if line.trim().is_empty() && !data.is_empty() {
760 // End of an event — try to parse it as a JSON-RPC message.
761 if let Ok(value) = serde_json::from_str::<Value>(&data)
762 && (value.get("result").is_some() || value.get("error").is_some())
763 {
764 return Some(value);
765 }
766 data.clear();
767 }
768 }
769 // Trailing event without a closing blank line.
770 if !data.is_empty()
771 && let Ok(value) = serde_json::from_str::<Value>(&data)
772 && (value.get("result").is_some() || value.get("error").is_some())
773 {
774 return Some(value);
775 }
776 None
777 }
778
779 // ── Status reporting (`/mcp`) ────────────────────────────────────────────────
780
781 /// Human-readable summary of configured MCP servers and their tools, for the
782 /// `/mcp` slash command.
783 pub fn status_summary() -> String {
784 let Some(mcp) = MCP.get() else {
785 return "MCP is not initialized.".to_string();
786 };
787 if mcp.servers.is_empty() {
788 return "No MCP servers configured. Add one in ~/.config/sigit/mcp.toml \
789 or .sigit/mcp.toml. See https://modelcontextprotocol.io."
790 .to_string();
791 }
792
793 let total_tools: usize = mcp.servers.iter().map(|s| s.tools.len()).sum();
794 let mut lines = vec![format!(
795 "{} MCP server(s), {total_tools} tool(s) available:",
796 mcp.servers.len()
797 )];
798 for server in &mcp.servers {
799 match &server.error {
800 Some(error) => lines.push(format!(
801 "- {} ({}) — unavailable: {error}",
802 server.name, server.url
803 )),
804 None => {
805 lines.push(format!(
806 "- {} ({}) — {} tool(s)",
807 server.name,
808 server.url,
809 server.tools.len()
810 ));
811 for tool in &server.tools {
812 lines.push(format!(" • {}", tool.full_name));
813 }
814 }
815 }
816 }
817 lines.join("\n")
818 }
819
820 #[cfg(test)]
821 mod tests {
822 use super::*;
823
824 #[test]
825 fn is_mcp_tool_detects_prefix() {
826 assert!(is_mcp_tool("mcp__sigit__search"));
827 assert!(!is_mcp_tool("read_file"));
828 assert!(!is_mcp_tool("skill"));
829 }
830
831 #[test]
832 fn sanitize_collapses_invalid_chars() {
833 assert_eq!(sanitize("github"), "github");
834 assert_eq!(sanitize("my server"), "my_server");
835 assert_eq!(sanitize("a.b/c:d"), "a_b_c_d");
836 assert_eq!(sanitize("keep-_ok9"), "keep-_ok9");
837 }
838
839 #[test]
840 fn parses_mcp_file_with_servers() {
841 let toml = r#"
842 official = false
843
844 [[server]]
845 name = "github"
846 url = "https://api.example.com/mcp"
847
848 [[server]]
849 name = "disabled-one"
850 url = "https://nope.example.com/mcp"
851 enabled = false
852
853 [server.headers]
854 Authorization = "Bearer xyz"
855 "#;
856 let parsed: McpFile = toml::from_str(toml).unwrap();
857 assert_eq!(parsed.official, Some(false));
858 assert_eq!(parsed.server.len(), 2);
859 assert_eq!(parsed.server[0].name, "github");
860 assert_eq!(parsed.server[1].enabled, Some(false));
861 assert_eq!(
862 parsed.server[1]
863 .headers
864 .get("Authorization")
865 .map(String::as_str),
866 Some("Bearer xyz")
867 );
868 }
869
870 #[test]
871 fn upsert_replaces_same_name() {
872 let mut defs = vec![ServerDef {
873 name: "a".into(),
874 url: "u1".into(),
875 headers: vec![],
876 }];
877 upsert(
878 &mut defs,
879 ServerDef {
880 name: "a".into(),
881 url: "u2".into(),
882 headers: vec![],
883 },
884 );
885 assert_eq!(defs.len(), 1);
886 assert_eq!(defs[0].url, "u2");
887 }
888
889 #[test]
890 fn parse_sse_extracts_jsonrpc_response() {
891 let body =
892 "event: message\ndata: {\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"ok\":true}}\n\n";
893 let value = parse_sse_response(body).expect("a message");
894 assert_eq!(value["result"]["ok"], json!(true));
895 }
896
897 #[test]
898 fn parse_sse_handles_no_trailing_blank_line() {
899 let body = "data: {\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{}}";
900 assert!(parse_sse_response(body).is_some());
901 }
902
903 #[test]
904 fn parse_sse_ignores_non_response_frames() {
905 // A lone notification (no result/error) shouldn't be mistaken for the response.
906 let body = "data: {\"jsonrpc\":\"2.0\",\"method\":\"ping\"}\n\n";
907 assert!(parse_sse_response(body).is_none());
908 }
909
910 #[test]
911 fn render_result_joins_text_blocks() {
912 let result = json!({
913 "content": [
914 { "type": "text", "text": "line one" },
915 { "type": "text", "text": "line two" }
916 ]
917 });
918 assert_eq!(render_tool_result(&result), "line one\nline two");
919 }
920
921 #[test]
922 fn render_result_marks_errors_and_non_text() {
923 let result = json!({
924 "isError": true,
925 "content": [
926 { "type": "text", "text": "boom" },
927 { "type": "image", "data": "..." }
928 ]
929 });
930 let rendered = render_tool_result(&result);
931 assert!(rendered.starts_with("Tool reported an error:"));
932 assert!(rendered.contains("boom"));
933 assert!(rendered.contains("[image content omitted]"));
934 }
935
936 #[test]
937 fn render_result_falls_back_to_structured_content() {
938 let result = json!({ "structuredContent": { "value": 42 } });
939 assert!(render_tool_result(&result).contains("42"));
940 }
941
942 #[test]
943 fn truncate_caps_long_output() {
944 let long = "x".repeat(RESULT_CHAR_LIMIT + 100);
945 let out = truncate(long);
946 assert!(out.contains("[output truncated"));
947 }
948
949 #[test]
950 fn tool_specs_empty_before_init() {
951 // Without init() the global is unset; this must not panic.
952 assert!(super::tool_specs().is_empty() || MCP.get().is_some());
953 }
954 }