linux.do

【有偿求助】怎么实现sse模式的mcp转发

最近研究MCP，发现remote模式的MCP有2种通信模式，streamableHttp 和 SSE，要对所有的MCP请求做统一转发管理，自己实现了streamableHttp的转发，但是SSE的搞不定，有佬帮忙看看吗，有偿求助

mcp测试工具：npx @modelcontextprotocol/inspector

streamableHttp MCP：npx @modelcontextprotocol/server-everything streamableHttp

sse MCP：npx @modelcontextprotocol/server-everything sse

下面是streamableHttp的转发代码

from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware
import httpx
from starlette.responses import StreamingResponse
import asyncio
import uvicorn

app = FastAPI()

# MCP服务器配置 - 所有远程URL都在这里集中管理，不对外暴露
MCP_SERVERS = {
    'microsoft': 'https://learn.microsoft.com/api/mcp',
    'everything': 'http://127.0.0.1:3001/mcp',
}

# 配置CORS中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 允许所有来源，生产环境应限制为特定域名
    allow_credentials=True,
    allow_methods=["*"],  # 允许所有HTTP方法
    allow_headers=["*"],  # 允许所有请求头
    expose_headers=["*"],  # 允许暴露所有响应头
)

# 通用的异步请求转发函数
async def forward_request_async(target_url: str, request: Request):
    # 构建请求头，移除不需要转发的头
    headers = dict(request.headers)
    headers.pop("host", None)
    headers.pop("content-length", None)
    headers.pop("connection", None)
    headers.pop("accept-encoding", None)

    # 获取请求体
    body = await request.body()

    # 创建httpx异步客户端
    async with httpx.AsyncClient(timeout=30.0) as client:
        try:
            # 发送请求到目标服务器
            response = await client.request(
                method=request.method,
                url=target_url,
                headers=headers,
                content=body,
                params=dict(request.query_params),
                follow_redirects=True
            )

            # 构建响应头，移除不能在响应中设置的头
            response_headers = dict(response.headers)
            response_headers.pop("content-encoding", None)
            response_headers.pop("transfer-encoding", None)
            response_headers.pop("connection", None)

            # 返回响应
            return StreamingResponse(
                content=response.aiter_bytes(),
                status_code=response.status_code,
                headers=response_headers
            )
        except httpx.RequestError as e:
            raise HTTPException(status_code=503, detail=f"无法连接到目标服务器: {str(e)}")

# MCP代理路由 - 处理/mcp/{server_id}路径
@app.api_route("/mcp/{server_id}", methods=["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS"])
async def mcp_proxy(server_id: str, request: Request):
    """MCP服务代理转发端点"""
    # 检查server_id是否在配置中
    if server_id not in MCP_SERVERS:
        raise HTTPException(
            status_code=404,
            detail=f"找不到配置的MCP服务器: {server_id}。可用的服务器: {list(MCP_SERVERS.keys())}"
        )

    # 获取目标URL
    target_url = MCP_SERVERS[server_id]

    # 对于OPTIONS请求，可以直接返回200，CORS中间件会添加必要的头
    if request.method == "OPTIONS":
        return {"status": "ok"}

    # 转发请求到目标MCP服务器
    return await forward_request_async(target_url, request)

if __name__ == "__main__":
    # 使用uvicorn运行FastAPI应用
    uvicorn.run(app, host="0.0.0.0", port=8000)

14 posts - 5 participants

Read full topic

via LINUX DO - 最新话题 (author: timeless)

23 views10:05