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
+}