component_shape_mcp/
stdio_smoke.rs

1use std::{
2    io::{BufRead as _, BufReader, Read as _, Write as _},
3    process::{Child, ChildStdin, ChildStdout, Command, ExitStatus, Stdio},
4    sync::{Arc, Mutex},
5    thread,
6    time::{Duration, Instant},
7};
8
9use serde_json::{Value, json};
10use thiserror::Error;
11
12use crate::MCP_PROTOCOL_VERSION;
13
14const DEFAULT_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(2);
15
16/// Error returned by the newline-delimited JSON-RPC stdio smoke client.
17#[derive(Debug, Error)]
18pub enum McpStdioSmokeError {
19    #[error("failed to spawn MCP stdio server: {source}")]
20    Spawn {
21        #[source]
22        source: std::io::Error,
23    },
24    #[error("spawned MCP stdio server has no {0}")]
25    MissingPipe(&'static str),
26    #[error("failed to write MCP request: {source}")]
27    Write {
28        #[source]
29        source: std::io::Error,
30    },
31    #[error("failed to flush MCP request: {source}")]
32    Flush {
33        #[source]
34        source: std::io::Error,
35    },
36    #[error("failed to read MCP response: {source}")]
37    Read {
38        #[source]
39        source: std::io::Error,
40    },
41    #[error("MCP stdio server closed stdout before responding to `{method}`{status}{stderr}")]
42    Eof {
43        method: String,
44        status: ProcessStatus,
45        stderr: StderrSnapshot,
46    },
47    #[error("MCP stdio server returned invalid JSON for `{method}`: {source}; line: {line}")]
48    InvalidJson {
49        method: String,
50        line: String,
51        #[source]
52        source: serde_json::Error,
53    },
54    #[error("MCP stdio server returned JSON-RPC error for `{method}`: {error}")]
55    Rpc { method: String, error: Value },
56    #[error("MCP stdio server response for `{method}` did not contain `result`: {response}")]
57    MissingResult { method: String, response: Value },
58    #[error(
59        "MCP stdio server response for `{method}` did not contain the expected id `{id}`: {response}"
60    )]
61    UnexpectedResponse {
62        method: String,
63        id: u64,
64        response: Value,
65    },
66    #[error("failed to wait for MCP stdio server shutdown: {source}")]
67    Wait {
68        #[source]
69        source: std::io::Error,
70    },
71    #[error("failed to kill MCP stdio server after shutdown timeout: {source}")]
72    Kill {
73        #[source]
74        source: std::io::Error,
75    },
76}
77
78/// Process status attached to stdio smoke failures.
79#[derive(Clone, Debug, Eq, PartialEq)]
80pub struct ProcessStatus(Option<String>);
81
82impl ProcessStatus {
83    fn running() -> Self {
84        Self(None)
85    }
86
87    fn exited(status: ExitStatus) -> Self {
88        Self(Some(status.to_string()))
89    }
90}
91
92impl std::fmt::Display for ProcessStatus {
93    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
94        match &self.0 {
95            Some(status) => write!(formatter, " (process status: {status})"),
96            None => Ok(()),
97        }
98    }
99}
100
101/// Captured stderr attached to stdio smoke failures.
102#[derive(Clone, Debug, Eq, PartialEq)]
103pub struct StderrSnapshot(String);
104
105impl StderrSnapshot {
106    fn empty() -> Self {
107        Self(String::new())
108    }
109
110    fn from_stderr(stderr: &Arc<Mutex<String>>) -> Self {
111        Self(
112            stderr
113                .lock()
114                .map(|stderr| stderr.clone())
115                .unwrap_or_else(|error| format!("stderr capture lock failed: {error}")),
116        )
117    }
118
119    pub fn as_str(&self) -> &str {
120        &self.0
121    }
122}
123
124impl std::fmt::Display for StderrSnapshot {
125    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
126        if self.0.trim().is_empty() {
127            return Ok(());
128        }
129        write!(formatter, "\nstderr:\n{}", self.0.trim_end())
130    }
131}
132
133/// Minimal newline-delimited JSON-RPC MCP client for subprocess smoke tests.
134///
135/// This intentionally exposes raw JSON protocol results so application-level
136/// smoke flows can assert the same field names seen by external MCP clients.
137pub struct McpStdioSmokeClient {
138    child: Child,
139    stdin: Option<ChildStdin>,
140    stdout: BufReader<ChildStdout>,
141    stderr: Arc<Mutex<String>>,
142    next_request_id: u64,
143}
144
145impl McpStdioSmokeClient {
146    /// Spawn an MCP stdio server process and connect to its stdin/stdout.
147    pub fn spawn(command: &mut Command) -> Result<Self, McpStdioSmokeError> {
148        let mut child = command
149            .stdin(Stdio::piped())
150            .stdout(Stdio::piped())
151            .stderr(Stdio::piped())
152            .spawn()
153            .map_err(|source| McpStdioSmokeError::Spawn { source })?;
154
155        let stdin = child
156            .stdin
157            .take()
158            .ok_or(McpStdioSmokeError::MissingPipe("stdin"))?;
159        let stdout = child
160            .stdout
161            .take()
162            .ok_or(McpStdioSmokeError::MissingPipe("stdout"))?;
163        let stderr = child
164            .stderr
165            .take()
166            .ok_or(McpStdioSmokeError::MissingPipe("stderr"))?;
167        let stderr = capture_stderr(stderr);
168
169        Ok(Self {
170            child,
171            stdin: Some(stdin),
172            stdout: BufReader::new(stdout),
173            stderr,
174            next_request_id: 1,
175        })
176    }
177
178    /// Discover the server's MCP 2026-07-28 capabilities and identity.
179    pub fn discover(&mut self) -> Result<Value, McpStdioSmokeError> {
180        self.request("server/discover", json!({}))
181    }
182
183    /// Call `tools/list`.
184    pub fn list_tools(&mut self) -> Result<Value, McpStdioSmokeError> {
185        self.request("tools/list", json!({}))
186    }
187
188    /// Call `resources/list`.
189    pub fn list_resources(&mut self) -> Result<Value, McpStdioSmokeError> {
190        self.request("resources/list", json!({}))
191    }
192
193    /// Call `resources/templates/list`.
194    pub fn list_resource_templates(&mut self) -> Result<Value, McpStdioSmokeError> {
195        self.request("resources/templates/list", json!({}))
196    }
197
198    /// Call `resources/read` for one concrete URI.
199    pub fn read_resource(&mut self, uri: &str) -> Result<Value, McpStdioSmokeError> {
200        self.request("resources/read", json!({ "uri": uri }))
201    }
202
203    /// Call `tools/call` for one tool with JSON object arguments.
204    pub fn call_tool(&mut self, name: &str, arguments: Value) -> Result<Value, McpStdioSmokeError> {
205        self.request(
206            "tools/call",
207            json!({
208                "name": name,
209                "arguments": arguments,
210            }),
211        )
212    }
213
214    /// Send a raw JSON-RPC request and return the response `result`.
215    pub fn request(
216        &mut self,
217        method: &str,
218        mut params: Value,
219    ) -> Result<Value, McpStdioSmokeError> {
220        let id = self.next_request_id;
221        self.next_request_id = self.next_request_id.saturating_add(1);
222        if let Some(params) = params.as_object_mut() {
223            params.insert(
224                "_meta".to_string(),
225                json!({
226                    "io.modelcontextprotocol/protocolVersion": MCP_PROTOCOL_VERSION,
227                    "io.modelcontextprotocol/clientCapabilities": {},
228                    "io.modelcontextprotocol/clientInfo": {
229                        "name": "component-shape-mcp-stdio-smoke",
230                        "version": env!("CARGO_PKG_VERSION"),
231                    },
232                }),
233            );
234        }
235        self.write_message(json!({
236            "jsonrpc": "2.0",
237            "id": id,
238            "method": method,
239            "params": params,
240        }))?;
241        self.read_response(method, id)
242    }
243
244    /// Close stdin, wait briefly for the server, and kill it if it remains alive.
245    pub fn shutdown(
246        &mut self,
247        timeout: Duration,
248    ) -> Result<Option<ExitStatus>, McpStdioSmokeError> {
249        self.stdin.take();
250        let deadline = Instant::now() + timeout;
251        loop {
252            match self
253                .child
254                .try_wait()
255                .map_err(|source| McpStdioSmokeError::Wait { source })?
256            {
257                Some(status) => return Ok(Some(status)),
258                None if Instant::now() >= deadline => {
259                    self.child
260                        .kill()
261                        .map_err(|source| McpStdioSmokeError::Kill { source })?;
262                    return self
263                        .child
264                        .wait()
265                        .map(Some)
266                        .map_err(|source| McpStdioSmokeError::Wait { source });
267                },
268                None => thread::sleep(Duration::from_millis(20)),
269            }
270        }
271    }
272
273    /// Return currently captured stderr text.
274    pub fn stderr(&self) -> StderrSnapshot {
275        StderrSnapshot::from_stderr(&self.stderr)
276    }
277
278    fn write_message(&mut self, message: Value) -> Result<(), McpStdioSmokeError> {
279        let stdin = self
280            .stdin
281            .as_mut()
282            .ok_or(McpStdioSmokeError::MissingPipe("stdin"))?;
283        serde_json::to_writer(&mut *stdin, &message).map_err(|source| {
284            McpStdioSmokeError::Write {
285                source: std::io::Error::other(source),
286            }
287        })?;
288        stdin
289            .write_all(b"\n")
290            .map_err(|source| McpStdioSmokeError::Write { source })?;
291        stdin
292            .flush()
293            .map_err(|source| McpStdioSmokeError::Flush { source })
294    }
295
296    fn read_response(&mut self, method: &str, id: u64) -> Result<Value, McpStdioSmokeError> {
297        let method = method.to_string();
298        loop {
299            let mut line = String::new();
300            let read = self
301                .stdout
302                .read_line(&mut line)
303                .map_err(|source| McpStdioSmokeError::Read { source })?;
304            if read == 0 {
305                let status = match self.child.try_wait() {
306                    Ok(Some(status)) => ProcessStatus::exited(status),
307                    Ok(None) | Err(_) => ProcessStatus::running(),
308                };
309                return Err(McpStdioSmokeError::Eof {
310                    method,
311                    status,
312                    stderr: self.stderr(),
313                });
314            }
315
316            let response = serde_json::from_str::<Value>(&line).map_err(|source| {
317                McpStdioSmokeError::InvalidJson {
318                    method: method.clone(),
319                    line: line.trim_end().to_string(),
320                    source,
321                }
322            })?;
323
324            if response.get("id").and_then(Value::as_u64) != Some(id) {
325                if response.get("id").is_none() {
326                    continue;
327                }
328                return Err(McpStdioSmokeError::UnexpectedResponse {
329                    method,
330                    id,
331                    response,
332                });
333            }
334
335            if let Some(error) = response.get("error") {
336                return Err(McpStdioSmokeError::Rpc {
337                    method,
338                    error: error.clone(),
339                });
340            }
341
342            return response
343                .get("result")
344                .cloned()
345                .ok_or(McpStdioSmokeError::MissingResult { method, response });
346        }
347    }
348}
349
350impl Drop for McpStdioSmokeClient {
351    fn drop(&mut self) {
352        let _ = self.shutdown(DEFAULT_SHUTDOWN_TIMEOUT);
353    }
354}
355
356/// Return a tool call's protocol-level `structuredContent`, if present.
357pub fn tool_call_structured_content(result: &Value) -> Option<&Value> {
358    result
359        .get("structuredContent")
360        .or_else(|| result.get("structured_content"))
361}
362
363fn capture_stderr(stderr: impl std::io::Read + Send + 'static) -> Arc<Mutex<String>> {
364    let output = Arc::new(Mutex::new(String::new()));
365    let output_for_thread = Arc::clone(&output);
366    thread::spawn(move || {
367        let mut stderr = BufReader::new(stderr);
368        let mut captured = String::new();
369        if stderr.read_to_string(&mut captured).is_ok()
370            && let Ok(mut output) = output_for_thread.lock()
371        {
372            *output = captured;
373        }
374    });
375    output
376}
377
378impl Default for StderrSnapshot {
379    fn default() -> Self {
380        Self::empty()
381    }
382}
383
384#[cfg(all(test, unix))]
385mod tests {
386    use std::{process::Command, time::Duration};
387
388    use serde_json::json;
389
390    use super::{
391        McpStdioSmokeClient, McpStdioSmokeError, ProcessStatus, StderrSnapshot,
392        tool_call_structured_content,
393    };
394
395    fn spawn_shell(script: &str) -> McpStdioSmokeClient {
396        McpStdioSmokeClient::spawn(Command::new("sh").arg("-c").arg(script))
397            .expect("shell smoke server should spawn")
398    }
399
400    #[test]
401    fn stdio_client_exercises_the_public_protocol_helpers() {
402        let mut client = spawn_shell(
403            r#"
404read discover
405printf '%s\n' '{"jsonrpc":"2.0","id":1,"result":{"resultType":"complete","supportedVersions":["2026-07-28"],"capabilities":{},"ttlMs":0,"cacheScope":"private","_meta":{"io.modelcontextprotocol/serverInfo":{"name":"example","version":"0.0.0"}}}}'
406read tools
407printf '%s\n' '{"jsonrpc":"2.0","id":2,"result":{"resultType":"complete","tools":[],"ttlMs":0,"cacheScope":"private"}}'
408read resources
409printf '%s\n' '{"jsonrpc":"2.0","id":3,"result":{"resultType":"complete","resources":[],"ttlMs":0,"cacheScope":"private"}}'
410read templates
411printf '%s\n' '{"jsonrpc":"2.0","id":4,"result":{"resultType":"complete","resourceTemplates":[],"ttlMs":0,"cacheScope":"private"}}'
412read resource
413printf '%s\n' '{"jsonrpc":"2.0","id":5,"result":{"resultType":"complete","contents":[],"ttlMs":0,"cacheScope":"private"}}'
414read tool
415printf '%s\n' '{"jsonrpc":"2.0","id":6,"result":{"resultType":"complete","structuredContent":{"ok":true}}}'
416"#,
417        );
418
419        assert_eq!(
420            client.discover().expect("discovery should succeed"),
421            json!({
422                "resultType": "complete",
423                "supportedVersions": ["2026-07-28"],
424                "capabilities": {},
425                "ttlMs": 0,
426                "cacheScope": "private",
427                "_meta": {
428                    "io.modelcontextprotocol/serverInfo": {
429                        "name": "example",
430                        "version": "0.0.0"
431                    }
432                }
433            })
434        );
435        assert_eq!(
436            client.list_tools().expect("tools/list should succeed"),
437            json!({
438                "resultType": "complete",
439                "tools": [],
440                "ttlMs": 0,
441                "cacheScope": "private"
442            })
443        );
444        assert_eq!(
445            client
446                .list_resources()
447                .expect("resources/list should succeed"),
448            json!({
449                "resultType": "complete",
450                "resources": [],
451                "ttlMs": 0,
452                "cacheScope": "private"
453            })
454        );
455        assert_eq!(
456            client
457                .list_resource_templates()
458                .expect("resources/templates/list should succeed"),
459            json!({
460                "resultType": "complete",
461                "resourceTemplates": [],
462                "ttlMs": 0,
463                "cacheScope": "private"
464            })
465        );
466        assert_eq!(
467            client
468                .read_resource("shape://example")
469                .expect("resources/read should succeed"),
470            json!({
471                "resultType": "complete",
472                "contents": [],
473                "ttlMs": 0,
474                "cacheScope": "private"
475            })
476        );
477        let result = client
478            .call_tool("shape_example", json!({ "value": 1 }))
479            .expect("tools/call should succeed");
480        assert_eq!(
481            tool_call_structured_content(&result),
482            Some(&json!({ "ok": true }))
483        );
484        assert!(
485            client
486                .shutdown(Duration::from_secs(1))
487                .expect("server should shut down")
488                .is_some()
489        );
490    }
491
492    #[test]
493    fn response_reader_skips_notifications_and_reports_protocol_errors() {
494        let cases = [
495            (
496                "read request; printf '%s\\n' '{\"jsonrpc\":\"2.0\",\"method\":\"notice\"}' '{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":7}'",
497                None,
498            ),
499            (
500                "read request; printf '%s\\n' '{\"jsonrpc\":\"2.0\",\"id\":9,\"result\":7}'",
501                Some("expected id `1`"),
502            ),
503            (
504                "read request; printf '%s\\n' '{\"jsonrpc\":\"2.0\",\"id\":1,\"error\":{\"code\":-1}}'",
505                Some("JSON-RPC error"),
506            ),
507            (
508                "read request; printf '%s\\n' '{\"jsonrpc\":\"2.0\",\"id\":1}'",
509                Some("did not contain `result`"),
510            ),
511            (
512                "read request; printf '%s\\n' 'not-json'",
513                Some("invalid JSON"),
514            ),
515        ];
516
517        for (script, expected_error) in cases {
518            let mut client = spawn_shell(script);
519            let result = client.request("example", json!({}));
520            match expected_error {
521                Some(expected_error) => assert!(
522                    result
523                        .expect_err("response should fail")
524                        .to_string()
525                        .contains(expected_error),
526                    "expected error containing `{expected_error}`"
527                ),
528                None => assert_eq!(result.expect("response should succeed"), json!(7)),
529            }
530        }
531    }
532
533    #[test]
534    fn client_reports_spawn_eof_and_closed_stdin_failures() {
535        let spawn_error =
536            match McpStdioSmokeClient::spawn(&mut Command::new("/definitely/not/a/program")) {
537                Ok(_) => panic!("invalid executable should fail"),
538                Err(error) => error,
539            };
540        assert!(matches!(spawn_error, McpStdioSmokeError::Spawn { .. }));
541
542        let mut exited = spawn_shell("read request; printf 'server detail\\n' >&2");
543        let eof = exited
544            .request("exited", json!({}))
545            .expect_err("closed stdout should fail");
546        assert!(matches!(eof, McpStdioSmokeError::Eof { .. }));
547        assert!(eof.to_string().contains("closed stdout"));
548
549        let mut closed = spawn_shell("read ignored");
550        closed
551            .shutdown(Duration::from_secs(1))
552            .expect("server should shut down when stdin closes");
553        assert!(matches!(
554            closed.request("example", json!({})),
555            Err(McpStdioSmokeError::MissingPipe("stdin"))
556        ));
557    }
558
559    #[test]
560    fn shutdown_kills_a_server_that_does_not_exit_after_stdin_closes() {
561        let mut client = spawn_shell("while :; do :; done");
562        let status = client
563            .shutdown(Duration::ZERO)
564            .expect("timed-out server should be killed")
565            .expect("killed server should return a status");
566
567        assert!(!status.success());
568    }
569
570    #[test]
571    fn diagnostics_format_status_stderr_and_structured_content() {
572        assert_eq!(ProcessStatus::running().to_string(), "");
573        assert_eq!(StderrSnapshot::default().to_string(), "");
574
575        let stderr = StderrSnapshot(" detail \n".to_string());
576        assert_eq!(stderr.as_str(), " detail \n");
577        assert_eq!(stderr.to_string(), "\nstderr:\n detail");
578
579        assert_eq!(
580            tool_call_structured_content(&json!({ "structured_content": 3 })),
581            Some(&json!(3))
582        );
583        assert_eq!(tool_call_structured_content(&json!({})), None);
584    }
585}
component_shape_mcp/stdio_smoke.rs

component_shape_mcp/
stdio_smoke.rs
