别让你的SSE MCP服务卡住！一行 requests.post 引发的性能危机

问题出现¶

在使用 Python 开发基于 MCP 的 SSE 服务时，如果不小心写了同步阻塞的代码，就非常容易踩进 IO 阻塞的大坑。来看下面这段示例代码：

from mcp.server.fastmcp import FastMCP
from pydantic import Field
import time
import requests

# 初始化mcp服务
mcp = FastMCP("hello-mcp-server")

@mcp.tool(name="url_get", description="访问某个url")
def get_url(url: str = Field(..., description="将要访问的url")):
    r = requests.get(url)
    return r.text

@mcp.tool(name="计算加法", description="计算加法")
def add(a: int = Field(..., description="第一个数"), b: int = Field(..., description="第二个数")):
    return a + b

def run():
    mcp.run(transport="sse")

if __name__ == "__main__":
    run()

上述代码中的工具函数是同步的，requests.get 是一个阻塞操作。当你把它运行在 transport="sse" 的环境下时，问题就来了，如果此时有多个客户端连接着这个sse服务器，那么之后的工具调用则会全部阻塞卡住。

🔎 为什么以前没出问题？¶

我们再来看一个**正确写法的对比案例**，以往开发中通常会使用异步函数：

# 初始化mcp服务
mcp = FastMCP("hello-mcp-server")

# 定义工具
@mcp.tool(name="高德天气查询助手", description=".....")
async def query_logistics(city: str = Field(description="要查询天气的城市名称")) -> str:
    ...
    async with httpx.AsyncClient(headers=headers) as client:
        response = await client.get(url)
        ...

这段代码使用了 async def，并且所有 IO 操作（如网络请求）都是通过 await 调用异步库完成的，整个事件循环是不会被阻塞的。

而我们一开始的代码里，工具函数是同步的，加上 requests.get，直接阻塞整个 worker。

🚨 会导致什么后果？¶

当你的服务以 SSE（Server-Sent Events）形式运行，并被多个客户端同时连接时：

[!bug] 卡死一旦有一个客户端调用了同步阻塞的函数（如 requests.get），所有其他客户端的请求也会被卡住！💥

这就是典型的 Python IO 阻塞问题 —— 单线程的事件循环被卡住了。

⚙️ 问题是怎么发生的？¶

来看 mcp.run(transport="sse") 的底层逻辑，它最终启动了一个 Uvicorn HTTP 服务：

async def run_sse_async(self) -> None:
    """Run the server using SSE transport."""
    starlette_app = self.sse_app()

    config = uvicorn.Config(
        starlette_app,
        host=self.settings.host,
        port=self.settings.port,
        log_level=self.settings.log_level.lower(),
    )
    server = uvicorn.Server(config)
    await server.serve()

而 uvicorn 在运行 FastAPI（或其他 ASGI 框架）时，是**异步驱动的**。如果你在异步环境中使用了同步阻塞的代码（如 time.sleep、requests.post 等），会阻塞整个事件循环线程，使得当前 worker 无法处理任何其他请求 🧨。

✅ 如何解决？¶

在编写 MCP SSE 工具函数时，遵循以下原则即可避免阻塞：

总是使用 async def 定义工具函数
将所有 IO 操作改为异步非阻塞的形式

具体方法如下：

✅ 使用原生异步方法（推荐）¶

将同步调用改为响应的异步库调用，如 requests 改为 httpx 异步实现：

from mcp.server.fastmcp import FastMCP
from pydantic import Field
import asyncio

# 初始化mcp服务
mcp = FastMCP("hello-mcp-server")

@mcp.tool(name="url_get", description="访问某个url")
async def get_url(url: str = Field(..., description="将要访问的url")):
    async with httpx.AsyncClient(headers=headers) as client:
        response = await client.get(url)
        return response.text

@mcp.tool(name="计算加法", description="计算加法")
async def add(a: int = Field(..., description="第一个数"), b: int = Field(..., description="第二个数")):
    return a + b

def run():
    mcp.run(transport="sse")

if __name__ == "__main__":
    run()

🔄 使用 `asyncio.to_thread()` 包装同步阻塞操作¶

适用于没有异步版本的库（如某些老旧的数据库或第三方 SDK）：

from mcp.server.fastmcp import FastMCP
from pydantic import Field
import time
import asyncio
import requests

# 初始化mcp服务
mcp = FastMCP("hello-mcp-server")

@mcp.tool(name="url_get", description="访问某个url")
async def get_url(url: str = Field(..., description="将要访问的url")):
    response = await asyncio.to_thread(requests.get, url)
    return response.text

@mcp.tool(name="计算加法", description="计算加法")
async def add(a: int = Field(..., description="第一个数"), b: int = Field(..., description="第二个数")):
    return a + b

def run():
    mcp.run(transport="sse")

if __name__ == "__main__":
    run()

stdio类型如何选择？¶

对于stdio类型的mcp服务，由于只有你个客户端连接，调用工具时是顺序调用，所以也就无所谓阻塞与非阻塞，但是出于未来扩展为sse类型mcp，还是建议写成异步async形式的函数。

🧾 总结¶

✅ 请务必牢记以下几点：

使用 Python 开发 SSE 类型的 MCP Server 时，一定要警惕 IO 阻塞问题
MCP SDK 底层基于 uvicorn，运行阻塞代码会**影响整个事件循环**
工具函数必须使用 async def，并确保所有阻塞操作为异步方式
异步方式的两种选择：
- 使用异步库（如 httpx, aiomysql, aiofiles 等）
- 使用 asyncio.to_thread() 将同步操作放入线程池

🧠 记住：在异步世界里，一次阻塞，满盘皆输。