audio_codebuddy/mcp-for-xiaozhi/mcp_server_http.py

"""
mcp_server_http.py — 小智 MCP 接入点（streamable-http 模式）

为什么需要这个文件：
  mcp_server_local.py 用 WebSocket（ws://），
  小智读长 audiobook 内容时 WebSocket 经常断流，体验差。
  这里改成 streamable-http（HTTP + JSON / SSE），传输长内容更稳定。

复用：
  audio_server.handle_request(req) 是纯函数，直接调它。

启动：
  .venv\Scripts\python.exe mcp_server_http.py
  监听 0.0.0.0:8766（HTTP）
"""
from __future__ import annotations

import json
import logging
import os

import uvicorn
from starlette.applications import Starlette
from starlette.middleware.cors import CORSMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse, Response
from starlette.routing import Route

import audio_server

# 复用 audio_server 的设置
API_BASE = audio_server.API_BASE
API_TIMEOUT = audio_server.API_TIMEOUT
logger = logging.getLogger("MCP_HTTP")


async def mcp_endpoint(request: Request) -> Response:
    """
    MCP streamable-http 端点。
    接收 JSON-RPC 请求，返回 JSON-RPC 响应。

    注意：streamable-http 协议允许两种传输：
    - application/json（普通 POST）
    - text/event-stream（SSE，用于流式响应）

    长内容场景下，建议一次性 POST 返回完整 JSON，
    避免 WebSocket 长连接断开问题。
    """
    try:
        body = await request.json()
    except Exception as e:
        return JSONResponse(
            {"jsonrpc": "2.0", "id": None, "error": {"code": -32700, "message": f"parse error: {e}"}},
            status_code=400,
        )

    logger.info("<- %s", json.dumps(body, ensure_ascii=False)[:300])

    # 复用 audio_server 的 JSON-RPC 处理
    resp = audio_server.handle_request(body)

    if resp is None:
        # notification 类型，不需要响应
        return Response(status_code=204)

    logger.info("-> %s", json.dumps(resp, ensure_ascii=False)[:300])

    # 返回 application/json 响应（不用 SSE，避免长内容分片问题）
    return JSONResponse(
        resp,
        headers={
            "Cache-Control": "no-cache",
            "X-MCP-Transport": "streamable-http",
        },
    )


async def health_check(request: Request) -> Response:
    """健康检查端点"""
    return JSONResponse({
        "status": "ok",
        "transport": "streamable-http",
        "api_base": API_BASE,
    })


# Starlette app
app = Starlette(
    debug=False,
    routes=[
        Route("/mcp", mcp_endpoint, methods=["POST"]),
        Route("/health", health_check, methods=["GET"]),
    ],
)

# CORS（小智可能从不同源访问）
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


def main():
    port = int(os.environ.get("MCP_HTTP_PORT", "8766"))
    logging.basicConfig(
        level="INFO",
        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    )
    logger.info(f"MCP HTTP Server 启动在 http://0.0.0.0:{port}/mcp")
    logger.info(f"API base = {API_BASE}")
    logger.info(f"transport = streamable-http (POST /mcp)")
    uvicorn.run(app, host="0.0.0.0", port=port, log_level="info")


if __name__ == "__main__":
    main()