myko_server/mcp/
server.rs

1//! MCP stdio server.
2//!
3//! Reads JSON-RPC requests from stdin, writes responses to stdout, and
4//! ferries tool execution through a remote Myko server via `MykoClient`.
5//!
6//! The in-process HTTP/WS MCP endpoints use the same dispatch core but a
7//! different [`Executor`]; see `mcp::http`, `mcp::ws`, and `mcp::dispatch`.
8
9use std::{
10    io::{self, BufRead, Write},
11    sync::Arc,
12};
13
14use hyphae::Watchable;
15use myko::{
16    client::{ConnectionStatus, MykoClient},
17    command::CommandRegistration,
18    query::QueryRegistration,
19    report::ReportRegistration,
20};
21use serde_json::Value;
22use tokio::sync::mpsc;
23
24use super::{
25    dispatch::{self, ServerInfo},
26    exec::Executor,
27    filter::{
28        CALLABLE_ALLOW_ENV, CALLABLE_DENY_ENV, ClientFilters, VISIBILITY_ALLOW_ENV,
29        VISIBILITY_DENY_ENV,
30    },
31    types::{McpError, McpRequest, McpResponse},
32};
33
34/// MCP Server for Myko stdio transport.
35///
36/// Automatically exposes all registered queries, reports, and commands
37/// through the MCP protocol.
38pub struct McpServer {
39    info: ServerInfo,
40}
41
42impl Default for McpServer {
43    fn default() -> Self {
44        Self::new()
45    }
46}
47
48impl McpServer {
49    /// Create a new MCP server with default settings.
50    pub fn new() -> Self {
51        Self {
52            info: ServerInfo::default(),
53        }
54    }
55
56    /// Create a new MCP server with custom name and version.
57    pub fn with_info(name: impl Into<String>, version: impl Into<String>) -> Self {
58        Self {
59            info: ServerInfo {
60                name: name.into(),
61                version: version.into(),
62                instructions: None,
63                operation_index: Arc::new(Vec::new()),
64            },
65        }
66    }
67
68    /// Set the optional `instructions` text returned in the MCP `initialize`
69    /// response. Surfaced to the model by the connecting client on connect.
70    pub fn with_instructions(mut self, instructions: impl Into<String>) -> Self {
71        self.info.instructions = Some(instructions.into());
72        self
73    }
74
75    /// Run the MCP server over stdio (blocking).
76    ///
77    /// Reads JSON-RPC requests from stdin and writes responses to stdout.
78    /// Logs go to stderr. Connects to a Myko WebSocket server via the
79    /// `MYKO_ADDRESS` env var (default `ws://localhost:5155`).
80    pub fn run_stdio(&self) -> io::Result<()> {
81        let rt = tokio::runtime::Runtime::new()?;
82        rt.block_on(async { self.run_stdio_async().await })
83    }
84
85    async fn run_stdio_async(&self) -> io::Result<()> {
86        let myko_address =
87            std::env::var("MYKO_ADDRESS").unwrap_or_else(|_| "ws://localhost:5155".to_string());
88
89        eprintln!("[myko-mcp] Connecting to Myko at {}", myko_address);
90
91        let client = Arc::new(MykoClient::new());
92        client.set_address(Some(myko_address));
93
94        let status_guard = client.connection_status().subscribe(move |signal| {
95            if let hyphae::Signal::Value(status) = signal {
96                match &**status {
97                    ConnectionStatus::Connected(addr) => {
98                        eprintln!("[myko-mcp] Connected to {}", addr)
99                    }
100                    ConnectionStatus::Connecting(addr) => {
101                        eprintln!("[myko-mcp] Connecting to {}", addr)
102                    }
103                    ConnectionStatus::Reconnecting(addr) => {
104                        eprintln!("[myko-mcp] Reconnecting to {}", addr)
105                    }
106                    ConnectionStatus::Idle => eprintln!("[myko-mcp] Idle"),
107                    ConnectionStatus::Disconnected => eprintln!("[myko-mcp] Disconnected"),
108                }
109            }
110        });
111        client.connection_status().own(status_guard);
112
113        let executor = Arc::new(Executor::Client(client));
114        let info = Arc::new(self.info.clone());
115
116        // Stdio MCP can't carry per-request headers, so the same three
117        // knobs as HTTP/WS come from env vars instead. Empty / unset =
118        // permissive default.
119        let filter = Arc::new(ClientFilters::from_strings(
120            std::env::var(VISIBILITY_ALLOW_ENV).ok().as_deref(),
121            std::env::var(VISIBILITY_DENY_ENV).ok().as_deref(),
122            std::env::var(CALLABLE_ALLOW_ENV).ok().as_deref(),
123            std::env::var(CALLABLE_DENY_ENV).ok().as_deref(),
124        ));
125
126        let (response_tx, mut response_rx) = mpsc::channel::<McpResponse>(32);
127
128        // Stdin reader thread: read lines, parse, dispatch.
129        let response_tx_clone = response_tx.clone();
130        let executor_clone = executor.clone();
131        let info_clone = info.clone();
132        let filter_clone = filter.clone();
133        std::thread::spawn(move || {
134            let stdin = io::stdin();
135            for line in stdin.lock().lines() {
136                let line = match line {
137                    Ok(l) => l,
138                    Err(e) => {
139                        eprintln!("[myko-mcp] stdin error: {}", e);
140                        continue;
141                    }
142                };
143                if line.is_empty() {
144                    continue;
145                }
146
147                let request: McpRequest = match serde_json::from_str(&line) {
148                    Ok(r) => r,
149                    Err(e) => {
150                        eprintln!("[myko-mcp] Parse error: {}", e);
151                        let response =
152                            McpResponse::error(Value::Null, McpError::parse_error(e.to_string()));
153                        let _ = response_tx_clone.blocking_send(response);
154                        continue;
155                    }
156                };
157
158                let response_tx = response_tx_clone.clone();
159                let executor = executor_clone.clone();
160                let info = info_clone.clone();
161                let filter = filter_clone.clone();
162                tokio::spawn(async move {
163                    if let Some(response) =
164                        dispatch::handle_request(request, &filter, &executor, &info).await
165                    {
166                        let _ = response_tx.send(response).await;
167                    }
168                });
169            }
170        });
171
172        // Write responses to stdout.
173        let mut stdout = io::stdout().lock();
174        while let Some(response) = response_rx.recv().await {
175            let json = serde_json::to_string(&response)?;
176            writeln!(stdout, "{}", json)?;
177            stdout.flush()?;
178        }
179
180        Ok(())
181    }
182
183    /// Get a summary of all registered items.
184    pub fn summary(&self) -> McpSummary {
185        let mut queries = Vec::new();
186        let mut reports = Vec::new();
187        let mut commands = Vec::new();
188
189        for reg in inventory::iter::<QueryRegistration> {
190            queries.push(QueryInfo {
191                query_id: reg.query_id.to_string(),
192                query_item_type: reg.query_item_type.to_string(),
193            });
194        }
195
196        for reg in inventory::iter::<ReportRegistration> {
197            reports.push(ReportInfo {
198                report_id: reg.report_id.to_string(),
199                output_type: reg.output_type.to_string(),
200            });
201        }
202
203        for reg in inventory::iter::<CommandRegistration> {
204            commands.push(CommandInfo {
205                command_id: reg.command_id.to_string(),
206                result_type: reg.result_type.to_string(),
207            });
208        }
209
210        McpSummary {
211            queries,
212            reports,
213            commands,
214        }
215    }
216}
217
218// ─────────────────────────────────────────────────────────────────────────────
219// Summary Types
220// ─────────────────────────────────────────────────────────────────────────────
221
222/// Summary of registered Myko items.
223#[derive(Debug, Clone)]
224pub struct McpSummary {
225    pub queries: Vec<QueryInfo>,
226    pub reports: Vec<ReportInfo>,
227    pub commands: Vec<CommandInfo>,
228}
229
230/// Query registration info.
231#[derive(Debug, Clone)]
232pub struct QueryInfo {
233    pub query_id: String,
234    pub query_item_type: String,
235}
236
237/// Report registration info.
238#[derive(Debug, Clone)]
239pub struct ReportInfo {
240    pub report_id: String,
241    pub output_type: String,
242}
243
244/// Command registration info.
245#[derive(Debug, Clone)]
246pub struct CommandInfo {
247    pub command_id: String,
248    pub result_type: String,
249}
250
251#[cfg(test)]
252mod tests {
253    use super::*;
254
255    #[test]
256    fn stdio_mcp_server_with_instructions_sets_field() {
257        let server = McpServer::with_info("test", "0.0.0").with_instructions("teach me");
258        assert_eq!(server.info.instructions.as_deref(), Some("teach me"));
259    }
260}
myko_server/mcp/server.rs

myko_server/mcp/
server.rs
