//! HTTP (streamable HTTP, 2025-06-18) transport 层。 //! //! 桌面进程内嵌 server 的传输层:POST /mcp 单响应 JSON;GET 405;notification 202 无 body。 //! 与 stdio transport 共享同一 dispatch/handler/Ctx,纯 df-mcp 内部实现,零 tauri 依赖。 //! //! 协议合规(streamable HTTP 2025-06-18): //! - POST /mcp → `application/json` 单响应(dispatch 层错误随 JSON-RPC 错误体回,HTTP 仍 200) //! - GET /mcp → 405 + `Allow: POST` //! - notification(id 缺省)→ 202 Accepted 无 body //! - batch 数组 body → 400 + INVALID_REQUEST(本 server 不支持 batch) //! - 非法 JSON body → 400 + PARSE_ERROR //! - 无状态 server:忽略 Mcp-Session-Id 头(合规) //! //! 桌面端注入方式:构造 [`McpHttpState`] 传入共享 db + on_tool_call 回调, //! [`build_router`] 得 Router,再 `axum::serve(listener, router)` 常驻监听。 use std::sync::Arc; use axum::{ body::Bytes, extract::State, http::{header, HeaderValue, StatusCode}, response::{IntoResponse, Response}, routing::post, Router, }; use df_storage::db::Database; use serde_json::{json, Value}; use crate::protocol::{McpMethod, Request, INVALID_REQUEST, PARSE_ERROR}; use crate::server::dispatch; use crate::tools::Ctx; /// 默认端口(桌面内嵌固定端口;可用 env `DEVFLOW_MCP_PORT` 覆盖)。 pub const DEFAULT_MCP_PORT: u16 = 18765; /// HTTP MCP 共享状态(axum State)。 pub struct McpHttpState { pub ctx: Ctx, pub read_only: bool, /// 成功 tools/call 回调(工具名)。桌面端注入 → emit df-data-changed;None=不回调。 pub on_tool_call: Option>, } impl McpHttpState { pub fn new( db: Arc, read_only: bool, on_tool_call: Option>, ) -> Self { Self { ctx: Ctx::new(db), read_only, on_tool_call, } } } /// 构造 axum Router(/mcp 单路由:POST 处理请求,GET 回 405)。 pub fn build_router(state: McpHttpState) -> Router { Router::new() .route("/mcp", post(post_mcp).get(get_mcp)) .with_state(Arc::new(state)) } /// 在已绑定 listener 上启动 HTTP server(永不返回直至 shutdown)。 /// /// 桌面端用法:bind 127.0.0.1:18765 → spawn(serve_on(listener, state)), /// 进程内常驻,多 Claude 会话 HTTP 直连同一进程。 pub async fn serve_on(listener: tokio::net::TcpListener, state: McpHttpState) -> anyhow::Result<()> { axum::serve(listener, build_router(state)) .await .map_err(|e| anyhow::anyhow!("axum::serve 失败: {e}")) } /// POST /mcp:解析 body → dispatch → JSON 单响应 / 202 notification。 async fn post_mcp(State(state): State>, body: Bytes) -> Response { match handle_body(&state, &body).await { Outcome::Json(status, value) => (status, axum::Json(value)).into_response(), Outcome::Accepted => StatusCode::ACCEPTED.into_response(), } } /// GET /mcp → 405 + `Allow: POST`(streamable HTTP 规范)。 async fn get_mcp() -> Response { let mut resp = StatusCode::METHOD_NOT_ALLOWED.into_response(); resp.headers_mut() .insert(header::ALLOW, HeaderValue::from_static("POST")); resp } /// 分发结果:单响应 JSON / notification 202。 enum Outcome { /// 单响应(HTTP 状态码 + JSON-RPC body) Json(StatusCode, Value), /// notification:202 Accepted,无 body Accepted, } /// 解析请求体并分发到共享 dispatch。 async fn handle_body(state: &McpHttpState, body: &[u8]) -> Outcome { // ① UTF-8 校验 let text = match std::str::from_utf8(body) { Ok(t) => t, Err(_) => { return Outcome::Json( StatusCode::BAD_REQUEST, rpc_error(None, PARSE_ERROR, "Parse error: 请求体不是合法 UTF-8"), ); } }; // ② 整体 JSON 解析:数组 = batch,本 server 不支持 let raw: Value = match serde_json::from_str(text) { Ok(v) => v, Err(_) => { return Outcome::Json( StatusCode::BAD_REQUEST, rpc_error(None, PARSE_ERROR, "Parse error: 非法 JSON"), ); } }; if raw.is_array() { return Outcome::Json( StatusCode::BAD_REQUEST, rpc_error(None, INVALID_REQUEST, "Invalid request: batch 请求不支持"), ); } // ③ 解析为 JSON-RPC Request(结构非法 → INVALID_REQUEST) let req: Request = match serde_json::from_value(raw) { Ok(r) => r, Err(_) => { return Outcome::Json( StatusCode::BAD_REQUEST, rpc_error(None, INVALID_REQUEST, "Invalid request"), ); } }; let method = McpMethod::from_request(&req); let is_notification = req.id.is_none(); // ④ notification(id 缺省)→ 202 Accepted 无 body if is_notification { match method { McpMethod::Initialized => { tracing::debug!(target: "df_mcp", "HTTP: 客户端 initialized 通知已收"); } _ => { tracing::debug!(target: "df_mcp", m = ?method, "HTTP: 忽略未识别通知"); } } return Outcome::Accepted; } // ⑤ 工具名预取(tools/call 成功回调用;method 随后 move 进 dispatch) let tool_name = match &method { McpMethod::ToolsCall { name, .. } => Some(name.clone()), _ => None, }; let resp = dispatch(&state.ctx, state.read_only, req.id.clone(), method).await; // ⑥ 成功 tools/call(JSON-RPC 无 error 且 result.isError≠true)→ on_tool_call 回调 // (桌面端据此刷新 GUI)。MC-5/业务失败(handler 返 isError=true)不触发, // 防假 df-data-changed;与 stdio fire_write_hook 共用 is_success_tool_call,两 transport 统一。 if crate::server::is_success_tool_call(&resp) { if let (Some(name), Some(cb)) = (tool_name, &state.on_tool_call) { cb(&name); } } let value = serde_json::to_value(&resp) .unwrap_or_else(|_| rpc_error(req.id, crate::protocol::INTERNAL_ERROR, "响应序列化失败")); Outcome::Json(StatusCode::OK, value) } /// 构造 JSON-RPC 2.0 错误响应体。 fn rpc_error(id: Option, code: i32, message: &str) -> Value { json!({ "jsonrpc": "2.0", "id": id, "error": { "code": code, "message": message } }) } // ============================================================ // 单测:tower oneshot 打 Router + open_in_memory DB // 覆盖:initialize / tools/list / notification 202 / GET 405 / 非法 JSON / batch / on_tool_call // ============================================================ #[cfg(test)] mod tests { use super::*; use axum::{ body::Body, http::{header, Request, StatusCode}, }; use std::sync::Mutex; use tower::ServiceExt; /// 构造内存 DB + McpHttpState(on_tool_call 可注入) async fn test_state( read_only: bool, on_tool_call: Option>, ) -> McpHttpState { let db = Arc::new(Database::open_in_memory().await.unwrap()); McpHttpState::new(db, read_only, on_tool_call) } /// oneshot 打 Router:method + body → (status, json body, headers) async fn send( router: Router, method: &str, body: &str, ) -> (StatusCode, Value, axum::http::HeaderMap) { let builder = Request::builder().uri("/mcp").method(method); let builder = if method == "GET" { builder } else { builder.header(header::CONTENT_TYPE, "application/json") }; let req = builder.body(Body::from(body.to_string())).unwrap(); let resp = router.clone().oneshot(req).await.unwrap(); let status = resp.status(); let headers = resp.headers().clone(); let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX) .await .unwrap(); let value: Value = if bytes.is_empty() { Value::Null } else { serde_json::from_slice(&bytes).unwrap_or(Value::Null) }; (status, value, headers) } #[tokio::test] async fn post_initialize_returns_server_info_and_capabilities() { let state = test_state(false, None).await; let router = build_router(state); let (status, v, _) = send( router, "POST", r#"{"jsonrpc":"2.0","id":1,"method":"initialize","params":{}}"#, ) .await; assert_eq!(status, StatusCode::OK); assert_eq!(v["jsonrpc"], "2.0"); assert_eq!(v["id"], 1); assert_eq!(v["result"]["protocolVersion"], crate::server::PROTOCOL_VERSION); assert_eq!(v["result"]["serverInfo"]["name"], crate::server::SERVER_NAME); assert!(v["result"]["capabilities"]["tools"].is_object()); } #[tokio::test] async fn post_tools_list_excludes_high_risk_by_default() { let state = test_state(false, None).await; let router = build_router(state); let (status, v, _) = send( router, "POST", r#"{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}"#, ) .await; assert_eq!(status, StatusCode::OK); let tools = v["result"]["tools"].as_array().unwrap(); let names: Vec<&str> = tools.iter().map(|t| t["name"].as_str().unwrap()).collect(); // 默认(非 read-only):Low + Medium 可见,High 不可见 assert!(names.contains(&"list_projects")); assert!(names.contains(&"create_project")); // Medium assert!(!names.contains(&"delete_project")); // High assert!(!names.contains(&"run_workflow")); // High } #[tokio::test] async fn post_notification_returns_202_no_body() { let state = test_state(false, None).await; let router = build_router(state); let (status, v, _) = send( router, "POST", r#"{"jsonrpc":"2.0","method":"notifications/initialized","params":{}}"#, ) .await; assert_eq!(status, StatusCode::ACCEPTED); assert_eq!(v, Value::Null); } #[tokio::test] async fn get_mcp_returns_405_allow_post() { let state = test_state(false, None).await; let router = build_router(state); let (status, _, headers) = send(router, "GET", "").await; assert_eq!(status, StatusCode::METHOD_NOT_ALLOWED); assert_eq!(headers.get(header::ALLOW).unwrap(), "POST"); } #[tokio::test] async fn post_invalid_json_returns_400_parse_error() { let state = test_state(false, None).await; let router = build_router(state); let (status, v, _) = send(router, "POST", "not json").await; assert_eq!(status, StatusCode::BAD_REQUEST); assert_eq!(v["error"]["code"], PARSE_ERROR); } #[tokio::test] async fn post_batch_returns_400_invalid_request() { let state = test_state(false, None).await; let router = build_router(state); let body = r#"[{"jsonrpc":"2.0","id":1,"method":"ping","params":{}}]"#; let (status, v, _) = send(router, "POST", body).await; assert_eq!(status, StatusCode::BAD_REQUEST); assert_eq!(v["error"]["code"], INVALID_REQUEST); } #[tokio::test] async fn tools_call_success_triggers_on_tool_call() { let calls: Arc>> = Arc::new(Mutex::new(Vec::new())); let calls_cb = calls.clone(); let on_tool_call: Option> = Some(Arc::new(move |name| { calls_cb.lock().unwrap().push(name.to_string()); })); let state = test_state(false, on_tool_call).await; let router = build_router(state); let body = r#"{"jsonrpc":"2.0","id":10,"method":"tools/call","params":{"name":"create_project","arguments":{"name":"HttpProj","description":"via http"}}}"#; let (status, v, _) = send(router, "POST", body).await; assert_eq!(status, StatusCode::OK); // 成功业务响应:isError 不置位 assert!( v["result"]["isError"].is_null() || v["result"]["isError"] == Value::Bool(false) ); assert!( calls.lock().unwrap().contains(&"create_project".to_string()), "成功 tools/call 应触发 on_tool_call,实际: {:?}", *calls.lock().unwrap() ); } }