faucet_cli/mcp/
mod.rs

1//! MCP (Model Context Protocol) server surface for faucet (issue #420).
2//!
3//! A transport-agnostic JSON-RPC 2.0 dispatcher ([`handle_message`]) plus the
4//! tool implementations ([`tools`]). Two front-ends drive it:
5//!
6//! - **stdio** — the `faucet mcp` subcommand ([`crate::commands::mcp`]), for
7//!   local agents (Claude Desktop / Code).
8//! - **Streamable HTTP** — the `/mcp` route mounted by `faucet serve --mcp`,
9//!   which inherits serve's bearer-auth + RBAC + audit.
10//!
11//! The dispatcher re-exposes existing faucet capabilities (list / schema /
12//! scaffold / validate / preview, and a gated `run_pipeline`) in the shape an
13//! LLM agent speaks; it re-implements no pipeline logic. Read-only tools are
14//! always available; the mutating `run_pipeline` tool appears only when the
15//! context is constructed with `allow_mutations = true` (the `--allow-mutations`
16//! flag, ANDed with the caller's RBAC scope on the HTTP transport).
17
18pub mod protocol;
19pub mod tools;
20
21use crate::auth_catalog::AuthCatalog;
22use protocol::*;
23use serde_json::{Value, json};
24
25/// Everything a tool handler needs, independent of transport.
26pub struct McpContext {
27    /// Shared auth providers (for `preview`/`run_pipeline` connector builds).
28    pub auth: AuthCatalog,
29    /// Whether mutating tools (`run_pipeline`) are exposed and callable.
30    pub allow_mutations: bool,
31}
32
33impl McpContext {
34    pub fn new(auth: AuthCatalog, allow_mutations: bool) -> Self {
35        Self {
36            auth,
37            allow_mutations,
38        }
39    }
40}
41
42/// Install a tracing subscriber that writes to **stderr** — stdout is reserved
43/// for the JSON-RPC message stream in `faucet mcp` (stdio) mode.
44pub fn install_stderr_tracing(level: &str) {
45    use tracing_subscriber::EnvFilter;
46    let filter = EnvFilter::try_new(level).unwrap_or_else(|_| EnvFilter::new("info"));
47    let _ = tracing_subscriber::fmt()
48        .with_env_filter(filter)
49        .with_writer(std::io::stderr)
50        .try_init();
51}
52
53/// Server identity reported in `initialize`.
54fn server_info() -> Value {
55    json!({ "name": "faucet", "version": env!("CARGO_PKG_VERSION") })
56}
57
58/// Handle one JSON-RPC message. Returns `Some(response_json)` for a request,
59/// or `None` for a notification (which gets no response). A malformed message
60/// yields a JSON-RPC parse/invalid-request error envelope.
61pub async fn handle_message(ctx: &McpContext, raw: &str) -> Option<String> {
62    let req: JsonRpcRequest = match serde_json::from_str(raw) {
63        Ok(r) => r,
64        Err(e) => {
65            return Some(render(error_no_id(
66                PARSE_ERROR,
67                format!("parse error: {e}"),
68            )));
69        }
70    };
71
72    // Notifications (no id) get no response, whatever the method.
73    if req.is_notification() {
74        return None;
75    }
76    let id = req.id.clone().unwrap_or(Value::Null);
77
78    let resp = match req.method.as_str() {
79        "initialize" => success(
80            id,
81            json!({
82                "protocolVersion": PROTOCOL_VERSION,
83                "capabilities": { "tools": {}, "resources": {} },
84                "serverInfo": server_info(),
85            }),
86        ),
87        "ping" => success(id, json!({})),
88        "tools/list" => {
89            let defs = tools::tool_defs(ctx);
90            success(id, json!({ "tools": defs }))
91        }
92        "tools/call" => match req.params.get("name").and_then(Value::as_str) {
93            Some(name) => {
94                let empty = json!({});
95                let args = req.params.get("arguments").unwrap_or(&empty);
96                let result = tools::call_tool(ctx, name, args).await;
97                success(id, result)
98            }
99            None => error(id, INVALID_PARAMS, "tools/call requires a 'name' parameter"),
100        },
101        "resources/list" => success(id, json!({ "resources": [] })),
102        "resources/read" => error(
103            id,
104            INVALID_PARAMS,
105            "no readable resources are exposed in this version",
106        ),
107        other => error(id, METHOD_NOT_FOUND, format!("unknown method '{other}'")),
108    };
109    Some(render(resp))
110}
111
112/// Drive an MCP session over any byte streams: read newline-delimited JSON-RPC
113/// from `reader`, write each response (one JSON object per line) to `writer`.
114/// Blank lines are skipped; notifications produce no output. Returns when the
115/// reader hits EOF. The `faucet mcp` stdio command wraps stdin/stdout with
116/// this; tests drive it with in-memory buffers.
117pub async fn serve_stdio<R, W>(ctx: &McpContext, reader: R, writer: &mut W) -> std::io::Result<()>
118where
119    R: tokio::io::AsyncBufRead + Unpin,
120    W: tokio::io::AsyncWrite + Unpin,
121{
122    use tokio::io::{AsyncBufReadExt, AsyncWriteExt};
123    let mut lines = reader.lines();
124    while let Some(line) = lines.next_line().await? {
125        let trimmed = line.trim();
126        if trimmed.is_empty() {
127            continue;
128        }
129        if let Some(response) = handle_message(ctx, trimmed).await {
130            writer.write_all(response.as_bytes()).await?;
131            writer.write_all(b"\n").await?;
132            writer.flush().await?;
133        }
134    }
135    Ok(())
136}
137
138fn render(v: Value) -> String {
139    serde_json::to_string(&v).unwrap_or_else(|_| {
140        r#"{"jsonrpc":"2.0","id":null,"error":{"code":-32603,"message":"failed to serialize response"}}"#.to_string()
141    })
142}
143
144#[cfg(test)]
145mod tests {
146    use super::*;
147
148    fn ctx() -> McpContext {
149        McpContext::new(
150            crate::auth_catalog::build_auth_catalog(None).unwrap(),
151            false,
152        )
153    }
154
155    async fn call(raw: &str) -> Value {
156        let s = handle_message(&ctx(), raw).await.expect("response");
157        serde_json::from_str(&s).unwrap()
158    }
159
160    #[tokio::test]
161    async fn initialize_reports_protocol_and_server() {
162        let v = call(r#"{"jsonrpc":"2.0","id":1,"method":"initialize","params":{}}"#).await;
163        assert_eq!(v["result"]["protocolVersion"], PROTOCOL_VERSION);
164        assert_eq!(v["result"]["serverInfo"]["name"], "faucet");
165        assert!(v["result"]["capabilities"]["tools"].is_object());
166    }
167
168    #[tokio::test]
169    async fn ping_ok() {
170        let v = call(r#"{"jsonrpc":"2.0","id":2,"method":"ping"}"#).await;
171        assert!(v["result"].is_object());
172    }
173
174    #[tokio::test]
175    async fn notification_gets_no_response() {
176        let out = handle_message(
177            &ctx(),
178            r#"{"jsonrpc":"2.0","method":"notifications/initialized"}"#,
179        )
180        .await;
181        assert!(out.is_none());
182    }
183
184    #[tokio::test]
185    async fn tools_list_has_readonly_and_hides_mutations() {
186        let v = call(r#"{"jsonrpc":"2.0","id":3,"method":"tools/list"}"#).await;
187        let names: Vec<String> = v["result"]["tools"]
188            .as_array()
189            .unwrap()
190            .iter()
191            .map(|t| t["name"].as_str().unwrap().to_string())
192            .collect();
193        assert!(names.contains(&"list_connectors".to_string()));
194        assert!(names.contains(&"validate_config".to_string()));
195        assert!(!names.contains(&"run_pipeline".to_string()));
196    }
197
198    #[tokio::test]
199    async fn tools_list_shows_mutations_when_allowed() {
200        let ctx = McpContext::new(crate::auth_catalog::build_auth_catalog(None).unwrap(), true);
201        let s = handle_message(&ctx, r#"{"jsonrpc":"2.0","id":3,"method":"tools/list"}"#)
202            .await
203            .unwrap();
204        assert!(s.contains("run_pipeline"));
205    }
206
207    #[tokio::test]
208    async fn tools_call_dispatches() {
209        let v = call(
210            r#"{"jsonrpc":"2.0","id":4,"method":"tools/call","params":{"name":"list_connectors","arguments":{"kind":"state"}}}"#,
211        )
212        .await;
213        assert_eq!(v["result"]["isError"], false);
214        assert!(
215            v["result"]["content"][0]["text"]
216                .as_str()
217                .unwrap()
218                .contains("state_stores")
219        );
220    }
221
222    #[tokio::test]
223    async fn tools_call_without_name_is_invalid_params() {
224        let v = call(r#"{"jsonrpc":"2.0","id":5,"method":"tools/call","params":{}}"#).await;
225        assert_eq!(v["error"]["code"], INVALID_PARAMS);
226    }
227
228    #[tokio::test]
229    async fn unknown_method_is_method_not_found() {
230        let v = call(r#"{"jsonrpc":"2.0","id":6,"method":"frobnicate"}"#).await;
231        assert_eq!(v["error"]["code"], METHOD_NOT_FOUND);
232    }
233
234    #[tokio::test]
235    async fn malformed_json_is_parse_error() {
236        let s = handle_message(&ctx(), "not json").await.unwrap();
237        let v: Value = serde_json::from_str(&s).unwrap();
238        assert_eq!(v["error"]["code"], PARSE_ERROR);
239    }
240
241    #[tokio::test]
242    async fn resources_list_is_empty() {
243        let v = call(r#"{"jsonrpc":"2.0","id":7,"method":"resources/list"}"#).await;
244        assert_eq!(v["result"]["resources"].as_array().unwrap().len(), 0);
245    }
246
247    #[tokio::test]
248    async fn resources_read_is_invalid_params() {
249        let v = call(r#"{"jsonrpc":"2.0","id":8,"method":"resources/read","params":{"uri":"x"}}"#)
250            .await;
251        assert_eq!(v["error"]["code"], INVALID_PARAMS);
252    }
253
254    #[test]
255    fn install_stderr_tracing_is_idempotent() {
256        // Just exercises the installer (try_init, so a second global subscriber
257        // is a no-op rather than a panic).
258        install_stderr_tracing("info");
259        install_stderr_tracing("debug");
260    }
261
262    #[tokio::test]
263    async fn serve_stdio_processes_lines_and_skips_blanks_and_notifications() {
264        use std::io::Cursor;
265        // A request, a blank line (skipped), a notification (no output), then a
266        // second request. Expect exactly two response lines.
267        let input = "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"ping\"}\n\
268                     \n\
269                     {\"jsonrpc\":\"2.0\",\"method\":\"notifications/initialized\"}\n\
270                     {\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"tools/list\"}\n";
271        let mut output: Vec<u8> = Vec::new();
272        let reader = tokio::io::BufReader::new(Cursor::new(input.as_bytes().to_vec()));
273        serve_stdio(&ctx(), reader, &mut output).await.unwrap();
274        let text = String::from_utf8(output).unwrap();
275        let lines: Vec<&str> = text.lines().collect();
276        assert_eq!(
277            lines.len(),
278            2,
279            "one response per request, none for blank/notification"
280        );
281        let first: Value = serde_json::from_str(lines[0]).unwrap();
282        assert_eq!(first["id"], 1);
283        let second: Value = serde_json::from_str(lines[1]).unwrap();
284        assert!(second["result"]["tools"].is_array());
285    }
286}
faucet_cli/mcp/mod.rs

faucet_cli/mcp/
mod.rs
