origin_mcp_stdio/
lib.rs

1//! MCP over stdio.
2//!
3//! The client starts the application as a child process and talks to it over its
4//! standard streams — no port to allocate, no token to exchange, and it works while
5//! the GUI is not running.
6//!
7//! # stdout belongs to the protocol
8//!
9//! Nothing else may write there. A single log line corrupts the stream and the client
10//! reports a parse error, which points nowhere near the actual cause. Configure
11//! logging with [`TelemetryConfig::for_stdout_protocol`] before serving.
12//!
13//! [`TelemetryConfig::for_stdout_protocol`]: https://docs.rs/origin-telemetry
14
15use origin_domain::{AppError, Result};
16use origin_mcp_core::McpServer;
17use tokio::io::{AsyncBufReadExt, AsyncWrite, AsyncWriteExt, BufReader};
18
19/// Serve until the client closes stdin.
20pub async fn serve(server: &McpServer) -> Result<()> {
21    let input = BufReader::new(tokio::io::stdin());
22    let output = tokio::io::stdout();
23
24    serve_streams(server, input, output).await
25}
26
27/// Serve over arbitrary streams. Used by the tests, which is the point of the split.
28pub async fn serve_streams<R, W>(
29    server: &McpServer,
30    input: BufReader<R>,
31    mut output: W,
32) -> Result<()>
33where
34    R: tokio::io::AsyncRead + Unpin,
35    W: AsyncWrite + Unpin,
36{
37    tracing::debug!("mcp stdio transport started");
38    let mut lines = input.lines();
39
40    loop {
41        let line = match lines.next_line().await {
42            Ok(Some(line)) => line,
43            // Client closed the stream: an ordinary end, not a failure.
44            Ok(None) => break,
45            Err(error) => {
46                return Err(AppError::internal(format!(
47                    "cannot read from stdin: {error}"
48                )));
49            }
50        };
51
52        if line.trim().is_empty() {
53            continue;
54        }
55
56        let Some(response) = server.handle_line(&line).await else {
57            // A notification. Answering one would itself be a protocol error.
58            continue;
59        };
60
61        let encoded = serde_json::to_string(&response)
62            .map_err(|error| AppError::internal(format!("cannot encode response: {error}")))?;
63
64        output
65            .write_all(encoded.as_bytes())
66            .await
67            .map_err(write_error)?;
68        output.write_all(b"\n").await.map_err(write_error)?;
69
70        // Flushed per message: the client is waiting for this answer before it sends
71        // the next request, so a buffered response deadlocks both sides.
72        output.flush().await.map_err(write_error)?;
73    }
74
75    tracing::debug!("mcp stdio transport stopped");
76    Ok(())
77}
78
79fn write_error(error: std::io::Error) -> AppError {
80    AppError::internal(format!("cannot write to stdout: {error}"))
81}
origin_mcp_stdio/lib.rs

origin_mcp_stdio/
lib.rs
