app/mcp/server.py

"""
MCP服务器实现 - 使用FastMCP和Streamable HTTP transport
"""

import logging
from typing import Optional
from starlette.applications import Starlette
from mcp.server import FastMCP
from mcp.server.streamable_http import TransportSecuritySettings

logger = logging.getLogger(__name__)

# 创建传输安全设置
# 禁用 DNS rebinding 保护（代理环境必需），允许所有 hosts
_transport_security = TransportSecuritySettings(
    enable_dns_rebinding_protection=False,
    allowed_hosts=["*"],
)

# 创建FastMCP实例
# streamable_http_path="/" 因为我们会挂载到 /mcp 路径
mcp = FastMCP(
    "Message Function Center",
    json_response=True,
    streamable_http_path="/",
    transport_security=_transport_security,
)

# 缓存 MCP ASGI 应用
_mcp_app: Optional[Starlette] = None


def get_mcp_app() -> Starlette:
    """获取 MCP ASGI 应用实例。

    必须在 lifespan 之前调用此方法来创建 session_manager，
    然后在 lifespan 中使用 session_manager.run() 启动 task group。

    Returns:
        MCP ASGI 应用实例
    """
    global _mcp_app
    if _mcp_app is None:
        _mcp_app = mcp.streamable_http_app()
        logger.info("MCP ASGI应用已创建")
    return _mcp_app