在 FastMCP 开发中，你经常会遇到这样的情况：@mcp.tool 工具函数想用 async def 享受异步的好处，但业务逻辑却依赖一个同步的 ORM 或 SDK。直接调用会阻塞事件循环，导致整个服务性能下降。

解决这个问题的利器，就是 asgiref.sync 模块。本文将深入讲解如何在 FastMCP 中用好它，写出既正确又高效的工具代码。

一、为什么需要 asgiref？问题从何而来
FastMCP 是异步的，其底层事件循环负责调度所有请求。如果在一个异步工具函数（async def）中直接调用同步阻塞代码（比如 time.sleep(5)、同步的数据库查询、requests.get()），会发生什么？整个事件循环会被卡住 5 秒，期间所有其他请求都无法处理。

这就是异步编程中的“阻塞事件循环”问题。解决方案是：将同步阻塞操作“转移”到事件循环之外的线程中执行，asgiref 就是为此设计的官方工具。

二、核心组件：sync_to_async
sync_to_async 的作用是：将同步函数包装成异步函数，并在线程池中执行，让异步代码可以安全地 await 它。

2.1 基础用法

from asgiref.sync import sync_to_async
from fastmcp import FastMCP

mcp = FastMCP("我的MCP服务")

# 同步业务逻辑
def get_user_sync(user_id: int):
    # 假设这里是同步的数据库查询
    time.sleep(1)
    return {"id": user_id, "name": "张三"}

#正确的异步工具
@mcp.tool
async def get_user(user_id: int) -> dict:
    # 将同步函数转为异步，安全调用
    result = await sync_to_async(get_user_sync)(user_id)
    return result

也可以直接用装饰器：

@sync_to_async
def get_user_sync(user_id: int):
    time.sleep(1)
    return {"id": user_id, "name": "张三"}

@mcp.tool
async def get_user(user_id: int) -> dict:
    return await get_user_sync(user_id)

2.2 thread_sensitive 参数（关键！）
这是 sync_to_async 最重要的参数，决定了同步代码在哪条线程中执行，直接影响性能。

thread_sensitive=True（默认值）——“安全模式”

所有标记为 thread_sensitive 的同步代码会在同一个共享线程中串行执行。这保证了线程局部状态（如数据库连接、Django ORM 的 thread-local）的一致性，避免连接混乱。

代价是：如果多个异步任务同时调用同一个 sync_to_async 函数，它们会排队执行，无法并发。例如 5 次各耗时 5 秒的操作，总耗时 25 秒。

thread_sensitive=False——“性能模式”

同步函数会被提交到线程池，可以并发执行。适合独立的、不依赖线程状态的阻塞操作，如加密、计算、独立的文件读写。5 次操作可以并发完成，总耗时约 5 秒。

如何在两者间选择？

调用 Django ORM、数据库连接、事务相关 的代码：必须用 thread_sensitive=True（默认），否则可能出错。

独立的 CPU 计算、无状态 I/O：可以设为 False，提升并发性能。

如果不确定，保持默认的 True 最安全。

三、另一个方向：async_to_sync
async_to_sync 的作用相反：在同步代码中调用异步函数。如果你有一段同步代码（比如 Celery 任务、Django 管理命令），需要调用一个异步 API，async_to_sync 可以把它转成同步调用。

from asgiref.sync import async_to_sync

# 假设这是已经写好的异步函数
async def fetch_api_data():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com")
        return response.json()

# 在同步函数中调用它
def sync_task():
    data = async_to_sync(fetch_api_data)()
    print(data)

在 FastMCP 中，由于服务本身是异步的，通常只需要用到 sync_to_async。async_to_sync 更适合“在同步环境下调用异步代码”的场景。

四、FastMCP 实战：最佳实践与常见陷阱
4.1 正确的做法

from fastmcp import FastMCP
from asgiref.sync import sync_to_async
from sqlmodel import Session, select
from database import engine
from models import Employee

mcp = FastMCP("薪福通MCP")

@mcp.tool
async def get_employee(employee_id: int) -> dict:
    """异步工具，安全调用同步数据库操作"""
    
    @sync_to_async
    def query_employee():
        with Session(engine) as session:
            employee = session.get(Employee, employee_id)
            if not employee:
                return {"error": "员工不存在"}
            return {"id": employee.id, "name": employee.name}
    
    return await query_employee()

@mcp.tool
async def get_employee_with_tasks(employee_id: int) -> dict:
    """多个同步操作组合，使用 thread_sensitive=False 提升性能（如果独立）"""
    
    @sync_to_async(thread_sensitive=False)
    def heavy_calculation():
        # 这里是独立的CPU密集型计算，不依赖线程状态
        import pandas as pd
        # ... 处理逻辑 ...
        return result
    
    # 对数据库操作仍用默认的 thread_sensitive=True
    @sync_to_async
    def get_employee_data():
        with Session(engine) as session:
            employee = session.get(Employee, employee_id)
            return employee
    
    # 并发执行两个独立操作
    import asyncio
    employee_data, calc_result = await asyncio.gather(
        get_employee_data(),
        heavy_calculation()
    )
    
    return {
        "employee": employee_data,
        "calculated": calc_result
    }

4.2 常见错误
错误1：直接调用同步函数

# 会阻塞事件循环！
@mcp.tool
async def get_employee(employee_id: int) -> dict:
    with Session(engine) as session:
        employee = session.get(Employee, employee_id)
        return {"id": employee.id}

错误2：在已有事件循环中使用 asyncio.run()

#  RuntimeError: asyncio.run() cannot be called from a running event loop
@mcp.tool
async def bad_tool():
    result = asyncio.run(some_sync_function())
    return result

FastMCP 运行在事件循环中，asyncio.run() 会尝试创建新事件循环，导致嵌套循环错误。正确做法是用 sync_to_async。

错误3：忘记 await

# result 是一个协程对象，没有执行
result = sync_to_async(blocking_func)()

4.3 性能对比速查

场景	总耗时	说明
直接同步调用	阻塞整个服务	❌ 不可行
sync_to_async 默认（thread_sensitive=True）	串行，多个请求排队执行	安全，但性能受限
sync_to_async(thread_sensitive=False)	并发执行	性能好，但有约束
纯异步代码	并发执行，性能最佳	需要原生异步驱动

五、总结：何时用哪种方式
场景 推荐方案
FastMCP 工具中调用同步 ORM/数据库 用 sync_to_async，保持默认 thread_sensitive=True
调用独立的 CPU/计算密集函数 @sync_to_async(thread_sensitive=False) 或 asyncio.to_thread()
同步代码中调用异步 API 使用 async_to_sync
代码简单，并发要求不高 干脆全部用同步工具（def tool()），不需要 async def
核心口诀：异步调同步，用 sync_to_async；同步调异步，用 async_to_sync；Django ORM 必须加 thread_sensitive=True