Skip to main content

max / makenotwork

kberg: add stdio MCP transport Adds server::stdio::serve (newline-delimited JSON-RPC over stdin/stdout), factoring JSON-RPC handling into a shared dispatch() used by both the HTTP handler and stdio. Lets a client launch the server as a subprocess (claude mcp add <name> -- <binary> --stdio). Adds 3 dispatch tests. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-13 19:23 UTC
Commit: 5bf1e633fde60676a685836f1157b469cedfe0eb
Parent: 67b93ca
3 files changed, +169 insertions, -18 deletions
M shared/kberg/Cargo.toml +1 -1
@@ -27,7 +27,7 @@ tower-http = { version = "0.6", optional = true, features = ["trace"] }
27 27
28 28 reqwest = { version = "0.12", optional = true, features = ["json"] }
29 29
30 - tokio = { version = "1", optional = true, features = ["rt-multi-thread", "macros", "net", "sync", "time"] }
30 + tokio = { version = "1", optional = true, features = ["rt-multi-thread", "macros", "net", "sync", "time", "io-std", "io-util"] }
31 31
32 32 [dev-dependencies]
33 33 tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "sync", "time"] }
M shared/kberg/src/server/mod.rs +120 -17
@@ -27,6 +27,7 @@ use crate::error::{Error, Result};
27 27 use crate::tool::{SurfaceProjection, ToolRegistry};
28 28
29 29 pub mod protocol;
30 + pub mod stdio;
30 31
31 32 use protocol::{
32 33 CallToolParams, InitializeResult, JsonRpcRequest, JsonRpcResponse, MCP_PROTOCOL_VERSION,
@@ -145,28 +146,42 @@ async fn handle_rpc(State(state): State<AppState>, Json(req): Json<Value>) -> im
145 146 }
146 147 };
147 148
148 - // Notifications (no `id`) get a 202 with no body.
149 - let Some(id) = req.id.clone() else {
150 - return StatusCode::ACCEPTED.into_response();
151 - };
149 + match dispatch(
150 + &state.registry,
151 + state.config.projection,
152 + state.config.grants.as_ref(),
153 + req,
154 + )
155 + .await
156 + {
157 + // Notifications (no `id`) get a 202 with no body.
158 + None => StatusCode::ACCEPTED.into_response(),
159 + Some(response) => (StatusCode::OK, Json(response)).into_response(),
160 + }
161 + }
152 162
163 + /// Transport-agnostic JSON-RPC dispatch. Handles `initialize`, `tools/list`,
164 + /// and `tools/call` against a registry. Returns `None` for a notification
165 + /// (a request with no `id`), which carries no response. Shared by the HTTP
166 + /// handler and the [`stdio`] transport.
167 + pub(crate) async fn dispatch(
168 + registry: &ToolRegistry,
169 + projection: SurfaceProjection,
170 + grants: Option<&HashSet<String>>,
171 + req: JsonRpcRequest,
172 + ) -> Option<JsonRpcResponse> {
173 + let id = req.id.clone()?;
153 174 let response = match req.method.as_str() {
154 175 "initialize" => JsonRpcResponse::ok(id, initialize_result()),
155 176 "tools/list" => {
156 - let specs = state.registry.specs_projected(state.config.projection);
177 + let specs = registry.specs_projected(projection);
157 178 JsonRpcResponse::ok(id, json!({ "tools": specs }))
158 179 }
159 180 "tools/call" => match serde_json::from_value::<CallToolParams>(req.params.unwrap_or(Value::Null)) {
160 - Ok(params) => {
161 - let result = state
162 - .registry
163 - .call(&params.name, params.arguments, state.config.grants.as_ref())
164 - .await;
165 - match result {
166 - Ok(r) => JsonRpcResponse::ok(id, serde_json::to_value(r).unwrap()),
167 - Err(e) => rpc_err_from(id, e),
168 - }
169 - }
181 + Ok(params) => match registry.call(&params.name, params.arguments, grants).await {
182 + Ok(r) => JsonRpcResponse::ok(id, serde_json::to_value(r).unwrap()),
183 + Err(e) => rpc_err_from(id, e),
184 + },
170 185 Err(e) => JsonRpcResponse::err(id, codes::INVALID_PARAMS, e.to_string()),
171 186 },
172 187 other => JsonRpcResponse::err(
@@ -175,8 +190,7 @@ async fn handle_rpc(State(state): State<AppState>, Json(req): Json<Value>) -> im
175 190 format!("unknown method: {other}"),
176 191 ),
177 192 };
178 -
179 - (StatusCode::OK, Json(response)).into_response()
193 + Some(response)
180 194 }
181 195
182 196 fn rpc_err_from(id: Value, e: Error) -> JsonRpcResponse {
@@ -202,3 +216,92 @@ fn initialize_result() -> Value {
202 216 })
203 217 .unwrap()
204 218 }
219 +
220 + #[cfg(test)]
221 + mod tests {
222 + use super::*;
223 + use crate::tool::{Tool, ToolCallResult, ToolKind};
224 + use async_trait::async_trait;
225 +
226 + struct Echo;
227 + #[async_trait]
228 + impl Tool for Echo {
229 + fn name(&self) -> &str {
230 + "echo"
231 + }
232 + fn description(&self) -> &str {
233 + "read echo"
234 + }
235 + fn kind(&self) -> ToolKind {
236 + ToolKind::Read
237 + }
238 + fn input_schema(&self) -> Value {
239 + json!({ "type": "object", "properties": {} })
240 + }
241 + async fn call(&self, _args: Value) -> crate::Result<ToolCallResult> {
242 + Ok(ToolCallResult::text("pong"))
243 + }
244 + }
245 +
246 + fn registry() -> ToolRegistry {
247 + let mut r = ToolRegistry::new();
248 + r.register(Echo);
249 + r
250 + }
251 +
252 + fn request(id: Option<Value>, method: &str, params: Value) -> JsonRpcRequest {
253 + serde_json::from_value(json!({
254 + "jsonrpc": "2.0",
255 + "id": id,
256 + "method": method,
257 + "params": params,
258 + }))
259 + .unwrap()
260 + }
261 +
262 + #[tokio::test]
263 + async fn dispatch_initialize_and_list_and_call() {
264 + let r = registry();
265 + let grants = HashSet::new();
266 +
267 + let init = dispatch(&r, SurfaceProjection::Full, Some(&grants), request(Some(json!(1)), "initialize", Value::Null))
268 + .await
269 + .unwrap();
270 + let v = serde_json::to_value(init).unwrap();
271 + assert_eq!(v["result"]["serverInfo"]["name"], "kberg");
272 +
273 + let list = dispatch(&r, SurfaceProjection::Full, Some(&grants), request(Some(json!(2)), "tools/list", Value::Null))
274 + .await
275 + .unwrap();
276 + let v = serde_json::to_value(list).unwrap();
277 + assert_eq!(v["result"]["tools"][0]["name"], "echo");
278 +
279 + let call = dispatch(
280 + &r,
281 + SurfaceProjection::Full,
282 + Some(&grants),
283 + request(Some(json!(3)), "tools/call", json!({ "name": "echo", "arguments": {} })),
284 + )
285 + .await
286 + .unwrap();
287 + let v = serde_json::to_value(call).unwrap();
288 + assert_eq!(v["result"]["content"][0]["text"], "pong");
289 + }
290 +
291 + #[tokio::test]
292 + async fn dispatch_notification_has_no_response() {
293 + let r = registry();
294 + let resp = dispatch(&r, SurfaceProjection::Full, None, request(None, "initialize", Value::Null)).await;
295 + assert!(resp.is_none());
296 + }
297 +
298 + #[tokio::test]
299 + async fn dispatch_unknown_method_errors() {
300 + let r = registry();
301 + let resp = dispatch(&r, SurfaceProjection::Full, None, request(Some(json!(9)), "bogus", Value::Null))
302 + .await
303 + .unwrap();
304 + let v = serde_json::to_value(resp).unwrap();
305 + assert_eq!(v["error"]["code"], codes::METHOD_NOT_FOUND);
306 + }
307 + }
A shared/kberg/src/server/stdio.rs +48
@@ -0,0 +1,48 @@
1 + //! stdio MCP transport: newline-delimited JSON-RPC over stdin/stdout.
2 + //!
3 + //! This is the transport an MCP client like Claude Code uses when it launches
4 + //! the server as a subprocess (`claude mcp add <name> -- <binary> --stdio`).
5 + //! One JSON-RPC message per line in on stdin, one per line out on stdout.
6 + //! Anything the process wants to log must go to **stderr** — stdout is the
7 + //! protocol channel and any stray byte there corrupts the stream.
8 + //!
9 + //! Reuses the same [`dispatch`](super::dispatch) as the HTTP server, so both
10 + //! transports expose an identical tool surface and capability gate.
11 +
12 + use serde_json::Value;
13 + use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
14 +
15 + use super::ServeConfig;
16 + use super::protocol::{JsonRpcResponse, codes};
17 + use crate::error::Result;
18 + use crate::tool::ToolRegistry;
19 +
20 + /// Serve the registry over stdio until stdin reaches EOF (the client closed the
21 + /// pipe). Blocks for the lifetime of the connection.
22 + pub async fn serve(registry: ToolRegistry, config: ServeConfig) -> Result<()> {
23 + let mut lines = BufReader::new(tokio::io::stdin()).lines();
24 + let mut stdout = tokio::io::stdout();
25 +
26 + while let Some(line) = lines.next_line().await? {
27 + let line = line.trim();
28 + if line.is_empty() {
29 + continue;
30 + }
31 + let response = match serde_json::from_str(line) {
32 + Ok(req) => super::dispatch(&registry, config.projection, config.grants.as_ref(), req).await,
33 + Err(e) => Some(JsonRpcResponse::err(
34 + Value::Null,
35 + codes::PARSE_ERROR,
36 + e.to_string(),
37 + )),
38 + };
39 + // Notifications (`dispatch` returned `None`) carry no reply.
40 + if let Some(response) = response {
41 + let mut buf = serde_json::to_vec(&response)?;
42 + buf.push(b'\n');
43 + stdout.write_all(&buf).await?;
44 + stdout.flush().await?;
45 + }
46 + }
47 + Ok(())
48 + }