faucet-cli 1.7.1

//! MCP (Model Context Protocol) server surface for faucet (issue #420).
//!
//! A transport-agnostic JSON-RPC 2.0 dispatcher ([`handle_message`]) plus the
//! tool implementations ([`tools`]). Two front-ends drive it:
//!
//! - **stdio** — the `faucet mcp` subcommand ([`crate::commands::mcp`]), for
//!   local agents (Claude Desktop / Code).
//! - **Streamable HTTP** — the `/mcp` route mounted by `faucet serve --mcp`,
//!   which inherits serve's bearer-auth + RBAC + audit.
//!
//! The dispatcher re-exposes existing faucet capabilities (list / schema /
//! scaffold / validate / preview, and a gated `run_pipeline`) in the shape an
//! LLM agent speaks; it re-implements no pipeline logic. Read-only tools are
//! always available; the mutating `run_pipeline` tool appears only when the
//! context is constructed with `allow_mutations = true` (the `--allow-mutations`
//! flag, ANDed with the caller's RBAC scope on the HTTP transport).

pub mod protocol;
pub mod tools;

use crate::auth_catalog::AuthCatalog;
use protocol::*;
use serde_json::{Value, json};

/// Everything a tool handler needs, independent of transport.
pub struct McpContext {
    /// Shared auth providers (for `preview`/`run_pipeline` connector builds).
    pub auth: AuthCatalog,
    /// Whether mutating tools (`run_pipeline`) are exposed and callable.
    pub allow_mutations: bool,
}

impl McpContext {
    pub fn new(auth: AuthCatalog, allow_mutations: bool) -> Self {
        Self {
            auth,
            allow_mutations,
        }
    }
}

/// Install a tracing subscriber that writes to **stderr** — stdout is reserved
/// for the JSON-RPC message stream in `faucet mcp` (stdio) mode.
pub fn install_stderr_tracing(level: &str) {
    use tracing_subscriber::EnvFilter;
    let filter = EnvFilter::try_new(level).unwrap_or_else(|_| EnvFilter::new("info"));
    let _ = tracing_subscriber::fmt()
        .with_env_filter(filter)
        .with_writer(std::io::stderr)
        .try_init();
}

/// Server identity reported in `initialize`.
fn server_info() -> Value {
    json!({ "name": "faucet", "version": env!("CARGO_PKG_VERSION") })
}

/// Handle one JSON-RPC message. Returns `Some(response_json)` for a request,
/// or `None` for a notification (which gets no response). A malformed message
/// yields a JSON-RPC parse/invalid-request error envelope.
pub async fn handle_message(ctx: &McpContext, raw: &str) -> Option<String> {
    let req: JsonRpcRequest = match serde_json::from_str(raw) {
        Ok(r) => r,
        Err(e) => {
            return Some(render(error_no_id(
                PARSE_ERROR,
                format!("parse error: {e}"),
            )));
        }
    };

    // Notifications (no id) get no response, whatever the method.
    if req.is_notification() {
        return None;
    }
    let id = req.id.clone().unwrap_or(Value::Null);

    let resp = match req.method.as_str() {
        "initialize" => success(
            id,
            json!({
                "protocolVersion": PROTOCOL_VERSION,
                "capabilities": { "tools": {}, "resources": {} },
                "serverInfo": server_info(),
            }),
        ),
        "ping" => success(id, json!({})),
        "tools/list" => {
            let defs = tools::tool_defs(ctx);
            success(id, json!({ "tools": defs }))
        }
        "tools/call" => match req.params.get("name").and_then(Value::as_str) {
            Some(name) => {
                let empty = json!({});
                let args = req.params.get("arguments").unwrap_or(&empty);
                let result = tools::call_tool(ctx, name, args).await;
                success(id, result)
            }
            None => error(id, INVALID_PARAMS, "tools/call requires a 'name' parameter"),
        },
        "resources/list" => success(id, json!({ "resources": [] })),
        "resources/read" => error(
            id,
            INVALID_PARAMS,
            "no readable resources are exposed in this version",
        ),
        other => error(id, METHOD_NOT_FOUND, format!("unknown method '{other}'")),
    };
    Some(render(resp))
}

/// Drive an MCP session over any byte streams: read newline-delimited JSON-RPC
/// from `reader`, write each response (one JSON object per line) to `writer`.
/// Blank lines are skipped; notifications produce no output. Returns when the
/// reader hits EOF. The `faucet mcp` stdio command wraps stdin/stdout with
/// this; tests drive it with in-memory buffers.
pub async fn serve_stdio<R, W>(ctx: &McpContext, reader: R, writer: &mut W) -> std::io::Result<()>
where
    R: tokio::io::AsyncBufRead + Unpin,
    W: tokio::io::AsyncWrite + Unpin,
{
    use tokio::io::{AsyncBufReadExt, AsyncWriteExt};
    let mut lines = reader.lines();
    while let Some(line) = lines.next_line().await? {
        let trimmed = line.trim();
        if trimmed.is_empty() {
            continue;
        }
        if let Some(response) = handle_message(ctx, trimmed).await {
            writer.write_all(response.as_bytes()).await?;
            writer.write_all(b"\n").await?;
            writer.flush().await?;
        }
    }
    Ok(())
}

fn render(v: Value) -> String {
    serde_json::to_string(&v).unwrap_or_else(|_| {
        r#"{"jsonrpc":"2.0","id":null,"error":{"code":-32603,"message":"failed to serialize response"}}"#.to_string()
    })
}

#[cfg(test)]
mod tests {
    use super::*;

    fn ctx() -> McpContext {
        McpContext::new(
            crate::auth_catalog::build_auth_catalog(None).unwrap(),
            false,
        )
    }

    async fn call(raw: &str) -> Value {
        let s = handle_message(&ctx(), raw).await.expect("response");
        serde_json::from_str(&s).unwrap()
    }

    #[tokio::test]
    async fn initialize_reports_protocol_and_server() {
        let v = call(r#"{"jsonrpc":"2.0","id":1,"method":"initialize","params":{}}"#).await;
        assert_eq!(v["result"]["protocolVersion"], PROTOCOL_VERSION);
        assert_eq!(v["result"]["serverInfo"]["name"], "faucet");
        assert!(v["result"]["capabilities"]["tools"].is_object());
    }

    #[tokio::test]
    async fn ping_ok() {
        let v = call(r#"{"jsonrpc":"2.0","id":2,"method":"ping"}"#).await;
        assert!(v["result"].is_object());
    }

    #[tokio::test]
    async fn notification_gets_no_response() {
        let out = handle_message(
            &ctx(),
            r#"{"jsonrpc":"2.0","method":"notifications/initialized"}"#,
        )
        .await;
        assert!(out.is_none());
    }

    #[tokio::test]
    async fn tools_list_has_readonly_and_hides_mutations() {
        let v = call(r#"{"jsonrpc":"2.0","id":3,"method":"tools/list"}"#).await;
        let names: Vec<String> = v["result"]["tools"]
            .as_array()
            .unwrap()
            .iter()
            .map(|t| t["name"].as_str().unwrap().to_string())
            .collect();
        assert!(names.contains(&"list_connectors".to_string()));
        assert!(names.contains(&"validate_config".to_string()));
        assert!(!names.contains(&"run_pipeline".to_string()));
    }

    #[tokio::test]
    async fn tools_list_shows_mutations_when_allowed() {
        let ctx = McpContext::new(crate::auth_catalog::build_auth_catalog(None).unwrap(), true);
        let s = handle_message(&ctx, r#"{"jsonrpc":"2.0","id":3,"method":"tools/list"}"#)
            .await
            .unwrap();
        assert!(s.contains("run_pipeline"));
    }

    #[tokio::test]
    async fn tools_call_dispatches() {
        let v = call(
            r#"{"jsonrpc":"2.0","id":4,"method":"tools/call","params":{"name":"list_connectors","arguments":{"kind":"state"}}}"#,
        )
        .await;
        assert_eq!(v["result"]["isError"], false);
        assert!(
            v["result"]["content"][0]["text"]
                .as_str()
                .unwrap()
                .contains("state_stores")
        );
    }

    #[tokio::test]
    async fn tools_call_without_name_is_invalid_params() {
        let v = call(r#"{"jsonrpc":"2.0","id":5,"method":"tools/call","params":{}}"#).await;
        assert_eq!(v["error"]["code"], INVALID_PARAMS);
    }

    #[tokio::test]
    async fn unknown_method_is_method_not_found() {
        let v = call(r#"{"jsonrpc":"2.0","id":6,"method":"frobnicate"}"#).await;
        assert_eq!(v["error"]["code"], METHOD_NOT_FOUND);
    }

    #[tokio::test]
    async fn malformed_json_is_parse_error() {
        let s = handle_message(&ctx(), "not json").await.unwrap();
        let v: Value = serde_json::from_str(&s).unwrap();
        assert_eq!(v["error"]["code"], PARSE_ERROR);
    }

    #[tokio::test]
    async fn resources_list_is_empty() {
        let v = call(r#"{"jsonrpc":"2.0","id":7,"method":"resources/list"}"#).await;
        assert_eq!(v["result"]["resources"].as_array().unwrap().len(), 0);
    }

    #[tokio::test]
    async fn resources_read_is_invalid_params() {
        let v = call(r#"{"jsonrpc":"2.0","id":8,"method":"resources/read","params":{"uri":"x"}}"#)
            .await;
        assert_eq!(v["error"]["code"], INVALID_PARAMS);
    }

    #[test]
    fn install_stderr_tracing_is_idempotent() {
        // Just exercises the installer (try_init, so a second global subscriber
        // is a no-op rather than a panic).
        install_stderr_tracing("info");
        install_stderr_tracing("debug");
    }

    #[tokio::test]
    async fn serve_stdio_processes_lines_and_skips_blanks_and_notifications() {
        use std::io::Cursor;
        // A request, a blank line (skipped), a notification (no output), then a
        // second request. Expect exactly two response lines.
        let input = "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"ping\"}\n\
                     \n\
                     {\"jsonrpc\":\"2.0\",\"method\":\"notifications/initialized\"}\n\
                     {\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"tools/list\"}\n";
        let mut output: Vec<u8> = Vec::new();
        let reader = tokio::io::BufReader::new(Cursor::new(input.as_bytes().to_vec()));
        serve_stdio(&ctx(), reader, &mut output).await.unwrap();
        let text = String::from_utf8(output).unwrap();
        let lines: Vec<&str> = text.lines().collect();
        assert_eq!(
            lines.len(),
            2,
            "one response per request, none for blank/notification"
        );
        let first: Value = serde_json::from_str(lines[0]).unwrap();
        assert_eq!(first["id"], 1);
        let second: Value = serde_json::from_str(lines[1]).unwrap();
        assert!(second["result"]["tools"].is_array());
    }
}