Python MCP 实战：构建 FastAPI 服务端与客户端示例

关键点:

• operation_id="get_data": 这个 ID 将作为 MCP 工具的名称。
• tags=["mcp"]: 我们通过标签（tag）来筛选哪些路由需要被 MCP 服务暴露。

2. 集成 FastApiMCP

接下来，我们实例化 FastApiMCP 并将其挂载到 FastAPI 应用上。

# main.py (续)
# 创建 FastApiMCP 实例并挂载
mcp = FastApiMCP(
    app,
    name="问题检索 mcp",
    description="测试描述",
    include_tags=["mcp"]  # 只暴露包含 "mcp" 标签的路由
)
mcp.mount_sse(mount_path="/sse")

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

关键点:

• include_tags=["mcp"]: 告诉 FastApiMCP 只扫描并注册那些 tags 列表中包含 "mcp" 的路由。
• mcp.mount_sse(mount_path="/sse"): 在指定的路径 (/sse) 上启用基于 SSE (Server-Sent Events) 的 MCP 通信端点。客户端将通过这个端点进行连接。

客户端实现 (test/mcp_client.py)

客户端使用 mcp 库来连接服务端，并执行工具调用。

1. 连接和初始化

客户端代码的核心是使用 sse_client 创建一个到服务器 /sse 端点的连接，并初始化一个 ClientSession。

# test/mcp_client.py
import asyncio
from mcp import ClientSession
from mcp.client.sse import sse_client

async def simple_client():
    """使用 mcp 库的简化客户端"""
    url = "http://127.0.0.1:8000/sse"  # 指向 SSE 挂载点
    try:
        async with sse_client(url) as (read, write):
            async with ClientSession(read, write) as session:
                # 1. 初始化会话（MCP 握手）
                await session.initialize()
                print("✅ MCP 会话初始化成功")
                # ... 后续操作
    except Exception as e:
        print(f"❌ 客户端连接失败：{e}")

if __name__ == "__main__":
    asyncio.run(simple_client())

关键点:

• sse_client(url): 创建一个异步上下文管理器，处理与 SSE 端点的连接。
• session.initialize(): 执行 MCP 握手，确保客户端和服务端可以正常通信。

2. 列出和调用工具

初始化成功后，我们就可以与服务端交互了，例如列出所有可用的工具，或者调用某个特定的工具。

# test/mcp_client.py (续，在 session 上下文内)
# 2. 列出可用工具
tools_result = await session.list_tools()
print("🛠️ 可用的工具:")
for tool in tools_result.tools:
    print(f" - {tool.name}: {tool.description}")

# 3. 调用工具
call_result = await session.call_tool("get_data", arguments={"question": "测试问题", "type": "test"})
if call_result.content:
    print(f"📝 调用结果：{call_result.content[0].text}")

关键点:

• session.list_tools(): 从服务端获取所有已注册工具的列表。
• session.call_tool("get_data", ...): 远程调用名为 get_data 的工具。
• arguments: 一个字典，包含了调用工具时需要传递的参数，这些参数对应于 main.py 中 get_data 函数的参数。

完整的 mcp_client.py 客户端代码如下：

import asyncio
from mcp import ClientSession
from mcp.client.sse import sse_client

async def simple_client():
    """使用官方 mcp 库的简化客户端"""
    # 配置 SSE 服务器参数
    url = "http://127.0.0.1:8000/sse"
    try:
        async with sse_client(url) as (read, write):
            async with ClientSession(read, write) as session:
                # 初始化会话（MCP 握手）
                await session.initialize()
                print("✅ MCP 会话初始化成功")
                
                # 列出可用工具
                tools_result = await session.list_tools()
                print("🛠️ 可用的工具:")
                for tool in tools_result.tools:
                    print(f" - {tool.name}: {tool.description}")
                
                # 调用 get_data 工具
                call_result = await session.call_tool("get_data", arguments={"question": "测试问题", "type": "test"})
                if call_result.content:
                    print(f"📝 调用结果：{call_result.content[0].text}")
    except Exception as e:
        print(f"❌ 客户端连接失败：{e}")

if __name__ == "__main__":
    asyncio.run(simple_client())

如何运行

1. 启动服务端

在项目根目录下，运行 main.py：

python main.py

你应该会看到 Uvicorn 启动服务的日志，表明服务正在 http://0.0.0.0:8000 上运行。

2. 运行客户端

打开另一个终端，运行 test/mcp_client.py：

python test/mcp_client.py

预期输出

如果一切顺利，你将在客户端的终端看到类似以下的输出：

✅ MCP 会话初始化成功
🛠️ 可用的工具:
 - get_data: 根据问题获取回复
📝 调用结果：这是关于 '测试问题' 的答案。

同时，在服务端的终端，你会看到 get_data 函数被触发时打印的信息：

收到问题：测试问题，类型：test

结论

本文通过一个简单的示例，展示了如何使用 fastapi-mcp 和 mcp 库在 Python 中构建 MCP 服务端和客户端。这种模式非常适合于需要将现有 Web API（特别是 FastAPI）快速封装成可远程调用的'工具集'的场景，为服务间通信提供了一种灵活且强大的替代方案。通过 operation_id 和 tags 的智能映射，集成过程变得非常直观和高效。