DevFlow/crates/df-mcp/src/server_http.rs

//! 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<Arc<dyn Fn(&str) + Send + Sync>>,
}

impl McpHttpState {
    pub fn new(
        db: Arc<Database>,
        read_only: bool,
        on_tool_call: Option<Arc<dyn Fn(&str) + Send + Sync>>,
    ) -> 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<Arc<McpHttpState>>, 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<Value>, 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<Arc<dyn Fn(&str) + Send + Sync>>,
    ) -> 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<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
        let calls_cb = calls.clone();
        let on_tool_call: Option<Arc<dyn Fn(&str) + Send + Sync>> = 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()
        );
    }
}