GPU-accelerated UI toolkit (Vulkan)
git clone https://git.lucas.co/cce-ui.git
src/mcp.rs (11.5K)
1 //! Minimal MCP (Model Context Protocol) server support for cce-ui apps.
2 //!
3 //! Implements the tools-only subset of the spec — `initialize`, `tools/list`,
4 //! `tools/call`, `ping` — over the Streamable HTTP transport (JSON-RPC 2.0 in
5 //! HTTP POST bodies), so any MCP client can inspect and drive a running app,
6 //! e.g. `claude mcp add --transport http <name> http://127.0.0.1:<port>/mcp`.
7 //! The server is stateless: no sessions, no SSE stream, no server-initiated
8 //! messages (a GET gets 405).
9 //!
10 //! An app declares its [`McpTool`]s and calls [`start_mcp_server`] with the
11 //! engine's calloop sender plus a constructor wrapping [`McpToolCall`] into
12 //! its `Application::Message`; each `tools/call` is executed on the app's
13 //! event loop and answered over the carried mpsc channel — the same bridge
14 //! pattern as cce-designer's HTTP automation API.
15
16 use std::io::{BufRead, BufReader, Read, Write};
17 use std::net::{TcpListener, TcpStream};
18 use std::sync::{mpsc, Arc};
19 use std::time::Duration;
20
21 use serde_json::{json, Value};
22
23 /// Protocol revisions this server accepts; the newest is offered when the
24 /// client requests anything else. The tools-only subset is identical across
25 /// all of them.
26 const PROTOCOL_VERSIONS: [&str; 3] = ["2025-06-18", "2025-03-26", "2024-11-05"];
27
28 /// How long a `tools/call` waits on the app's event loop before failing.
29 const CALL_TIMEOUT: Duration = Duration::from_secs(30);
30
31 /// A tool the app exposes over MCP.
32 #[derive(Debug, Clone)]
33 pub struct McpTool {
34 pub name: String,
35 pub description: String,
36 /// JSON Schema for the tool's arguments (the spec's `inputSchema`).
37 pub input_schema: Value,
38 }
39
40 /// A `tools/call` in flight: delivered to the app's event loop wrapped in its
41 /// `Application::Message`; the handler sends the outcome back over `reply`.
42 /// An `Ok` value becomes the result's text content (strings verbatim, other
43 /// JSON pretty-printed); an `Err` becomes an `isError` tool result.
44 #[derive(Debug, Clone)]
45 pub struct McpToolCall {
46 pub name: String,
47 pub arguments: Value,
48 pub reply: mpsc::Sender<Result<Value, String>>,
49 }
50
51 /// Spawn the MCP server on `127.0.0.1:port` (one thread per connection,
52 /// mirroring the raw-HTTP style of cce-designer's api.rs). `wrap` lifts a
53 /// tool call into the app's message type for delivery over `sender`.
54 pub fn start_mcp_server<M, F>(
55 server_name: &str,
56 port: u16,
57 tools: Vec<McpTool>,
58 sender: calloop::channel::Sender<M>,
59 wrap: F,
60 ) where
61 M: Send + 'static,
62 F: Fn(McpToolCall) -> M + Send + Sync + 'static,
63 {
64 let server_name = server_name.to_string();
65 std::thread::spawn(move || {
66 let listener = match TcpListener::bind(("127.0.0.1", port)) {
67 Ok(l) => l,
68 Err(e) => {
69 eprintln!("Failed to bind MCP server to port {port}: {e:?}");
70 return;
71 }
72 };
73 println!("MCP server '{server_name}' listening on http://127.0.0.1:{port}");
74
75 let shared = Arc::new((server_name, tools, sender, wrap));
76 for stream in listener.incoming() {
77 let stream = match stream {
78 Ok(s) => s,
79 Err(_) => continue,
80 };
81 let shared = Arc::clone(&shared);
82 std::thread::spawn(move || {
83 let (server_name, tools, sender, wrap) = &*shared;
84 handle_connection(stream, server_name, tools, &|name, arguments| {
85 let (tx, rx) = mpsc::channel();
86 let call = McpToolCall { name: name.to_string(), arguments, reply: tx };
87 sender
88 .send(wrap(call))
89 .map_err(|_| "app event loop is gone".to_string())?;
90 rx.recv_timeout(CALL_TIMEOUT)
91 .map_err(|_| "timed out waiting for the app".to_string())?
92 });
93 });
94 }
95 });
96 }
97
98 fn handle_connection<F>(stream: TcpStream, server_name: &str, tools: &[McpTool], call_tool: &F)
99 where
100 F: Fn(&str, Value) -> Result<Value, String>,
101 {
102 let mut write_stream = match stream.try_clone() {
103 Ok(s) => s,
104 Err(_) => return,
105 };
106 let mut reader = BufReader::new(stream);
107 let mut request_line = String::new();
108 if reader.read_line(&mut request_line).is_err() {
109 return;
110 }
111
112 let mut content_length = 0usize;
113 loop {
114 let mut line = String::new();
115 if reader.read_line(&mut line).is_err() || line == "\r\n" || line == "\n" || line.is_empty()
116 {
117 break;
118 }
119 let lower = line.to_lowercase();
120 if let Some(rest) = lower.strip_prefix("content-length:") {
121 if let Ok(len) = rest.trim().parse::<usize>() {
122 content_length = len;
123 }
124 }
125 }
126
127 if !request_line.starts_with("POST ") {
128 // Stateless server: no SSE stream (GET) or session teardown (DELETE).
129 let _ = write_stream.write_all(
130 b"HTTP/1.1 405 Method Not Allowed\r\nAllow: POST\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
131 );
132 return;
133 }
134
135 let mut body = vec![0; content_length];
136 if reader.read_exact(&mut body).is_err() {
137 return;
138 }
139
140 let response = match serde_json::from_slice::<Value>(&body) {
141 Ok(req) => handle_jsonrpc(&req, server_name, tools, call_tool),
142 Err(_) => Some(error_response(Value::Null, -32700, "parse error")),
143 };
144 let http = match response {
145 Some(resp) => {
146 let body = resp.to_string();
147 format!(
148 "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
149 body.len(),
150 body
151 )
152 }
153 // Notifications get no JSON-RPC response, just an HTTP ack.
154 None => "HTTP/1.1 202 Accepted\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".to_string(),
155 };
156 let _ = write_stream.write_all(http.as_bytes());
157 let _ = write_stream.flush();
158 }
159
160 /// Dispatch one JSON-RPC message. Returns `None` for notifications (no id).
161 fn handle_jsonrpc<F>(req: &Value, server_name: &str, tools: &[McpTool], call_tool: &F) -> Option<Value>
162 where
163 F: Fn(&str, Value) -> Result<Value, String>,
164 {
165 let method = req.get("method").and_then(Value::as_str).unwrap_or("");
166 let id = match req.get("id") {
167 Some(id) if !id.is_null() => id.clone(),
168 _ => return None,
169 };
170
171 let result = match method {
172 "initialize" => {
173 let requested = req
174 .pointer("/params/protocolVersion")
175 .and_then(Value::as_str)
176 .unwrap_or("");
177 let version = if PROTOCOL_VERSIONS.contains(&requested) {
178 requested
179 } else {
180 PROTOCOL_VERSIONS[0]
181 };
182 json!({
183 "protocolVersion": version,
184 "capabilities": { "tools": {} },
185 "serverInfo": { "name": server_name, "version": env!("CARGO_PKG_VERSION") },
186 })
187 }
188 "ping" => json!({}),
189 "tools/list" => json!({
190 "tools": tools.iter().map(|t| json!({
191 "name": t.name,
192 "description": t.description,
193 "inputSchema": t.input_schema,
194 })).collect::<Vec<_>>(),
195 }),
196 "tools/call" => {
197 let name = req
198 .pointer("/params/name")
199 .and_then(Value::as_str)
200 .unwrap_or("");
201 if !tools.iter().any(|t| t.name == name) {
202 return Some(error_response(id, -32602, &format!("unknown tool: {name}")));
203 }
204 let arguments = req
205 .pointer("/params/arguments")
206 .cloned()
207 .unwrap_or_else(|| json!({}));
208 match call_tool(name, arguments) {
209 Ok(value) => {
210 let text = match value {
211 Value::String(s) => s,
212 other => serde_json::to_string_pretty(&other).unwrap_or_default(),
213 };
214 json!({ "content": [{ "type": "text", "text": text }], "isError": false })
215 }
216 Err(e) => json!({ "content": [{ "type": "text", "text": e }], "isError": true }),
217 }
218 }
219 _ => return Some(error_response(id, -32601, &format!("method not found: {method}"))),
220 };
221 Some(json!({ "jsonrpc": "2.0", "id": id, "result": result }))
222 }
223
224 fn error_response(id: Value, code: i64, message: &str) -> Value {
225 json!({ "jsonrpc": "2.0", "id": id, "error": { "code": code, "message": message } })
226 }
227
228 #[cfg(test)]
229 mod tests {
230 use super::*;
231
232 fn tools() -> Vec<McpTool> {
233 vec![McpTool {
234 name: "echo".to_string(),
235 description: "Echo the arguments back".to_string(),
236 input_schema: json!({ "type": "object" }),
237 }]
238 }
239
240 fn no_calls(_: &str, _: Value) -> Result<Value, String> {
241 panic!("no tool call expected");
242 }
243
244 #[test]
245 fn initialize_negotiates_protocol_version() {
246 let req = json!({
247 "jsonrpc": "2.0", "id": 1, "method": "initialize",
248 "params": { "protocolVersion": "2025-03-26" }
249 });
250 let resp = handle_jsonrpc(&req, "test", &tools(), &no_calls).unwrap();
251 assert_eq!(resp["result"]["protocolVersion"], "2025-03-26");
252 assert!(resp["result"]["capabilities"]["tools"].is_object());
253
254 // Unknown requested version falls back to our newest.
255 let req = json!({
256 "jsonrpc": "2.0", "id": 2, "method": "initialize",
257 "params": { "protocolVersion": "2099-01-01" }
258 });
259 let resp = handle_jsonrpc(&req, "test", &tools(), &no_calls).unwrap();
260 assert_eq!(resp["result"]["protocolVersion"], PROTOCOL_VERSIONS[0]);
261 }
262
263 #[test]
264 fn notifications_get_no_response() {
265 let req = json!({ "jsonrpc": "2.0", "method": "notifications/initialized" });
266 assert!(handle_jsonrpc(&req, "test", &tools(), &no_calls).is_none());
267 }
268
269 #[test]
270 fn tools_list_reports_declared_tools() {
271 let req = json!({ "jsonrpc": "2.0", "id": 3, "method": "tools/list" });
272 let resp = handle_jsonrpc(&req, "test", &tools(), &no_calls).unwrap();
273 assert_eq!(resp["result"]["tools"][0]["name"], "echo");
274 assert!(resp["result"]["tools"][0]["inputSchema"].is_object());
275 }
276
277 #[test]
278 fn tools_call_wraps_ok_and_err_results() {
279 let req = json!({
280 "jsonrpc": "2.0", "id": 4, "method": "tools/call",
281 "params": { "name": "echo", "arguments": { "x": 1 } }
282 });
283 let resp = handle_jsonrpc(&req, "test", &tools(), &|name, args| {
284 assert_eq!(name, "echo");
285 Ok(args)
286 })
287 .unwrap();
288 assert_eq!(resp["result"]["isError"], false);
289 assert!(resp["result"]["content"][0]["text"].as_str().unwrap().contains("\"x\": 1"));
290
291 let resp = handle_jsonrpc(&req, "test", &tools(), &|_, _| Err("boom".to_string())).unwrap();
292 assert_eq!(resp["result"]["isError"], true);
293 assert_eq!(resp["result"]["content"][0]["text"], "boom");
294 }
295
296 #[test]
297 fn unknown_tool_and_method_are_protocol_errors() {
298 let req = json!({
299 "jsonrpc": "2.0", "id": 5, "method": "tools/call",
300 "params": { "name": "nope" }
301 });
302 let resp = handle_jsonrpc(&req, "test", &tools(), &no_calls).unwrap();
303 assert_eq!(resp["error"]["code"], -32602);
304
305 let req = json!({ "jsonrpc": "2.0", "id": 6, "method": "resources/list" });
306 let resp = handle_jsonrpc(&req, "test", &tools(), &no_calls).unwrap();
307 assert_eq!(resp["error"]["code"], -32601);
308 }
309 }