Python MCP实战：构建 FastAPI 服务端与客户端示例&MCP客户端调用
引言
在现代微服务架构中，服务间的通信协议选择至关重要。除了常见的 RESTful API、gRPC 等，MCP（Message-oriented Communication Protocol）作为一种面向消息的通信协议，也逐渐在特定场景中展现出其优势。本文将通过一个具体的 Python 示例，演示如何基于 fastapi-mcp 和 mcp 库，构建一个 MCP 服务端和客户端，并实现工具（Tool）的远程调用。
服务端将使用 FastAPI 框架，通过 fastapi-mcp 库将一个 API endpoint 暴露为 MCP 工具。客户端则会演示如何连接到 MCP 服务，列出可用的工具，并远程调用它。
核心组件:
- 服务端 (
main.py): 一个基于 FastAPI 的 Web 应用，提供 MCP 服务。 - 客户端 (
test/mcp_client.py): 一个 Python 脚本，用于连接 MCP 服务并与之交互。
服务端实现 (main.py)
我们的服务端基于 FastAPI 构建。核心是利用 fastapi-mcp 库将指定的 FastAPI 路由转换成 MCP 可以识别和调用的“工具”。
1. 定义 FastAPI 路由
首先，我们定义一个标准的 FastAPI 路由，这个路由将是我们要通过 MCP 暴露的功能。在这个例子中，我们创建了一个名为 get_date 的异步函数，它接收 question 和 type 两个参数，并返回一个答案。
# main.pyfrom fastapi import FastAPI from fastapi_mcp import FastApiMCP import uvicorn # 创建FastAPI应用 app = FastAPI( title="GRAPH RAG API", description="高效的web数据管理接口", version="0.1.0",)# ... (其他FastAPI配置，如CORS、异常处理等)@app.get("/answer", operation_id="get_data", tags=["mcp"], summary="根据问题获取答案")asyncdefget_data(question:str,type:str):# 在实际应用中，这里会调用服务层处理业务逻辑# chat_service = get_chat_service()# answer = await chat_service.get_answer_mcp(question,type)print(f"收到问题: {question}, 类型: {type}")returnf"这是关于 '{question}' 的答案。"# ...关键点:
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.pyimport asyncio from mcp import ClientSession from mcp.client.sse import sse_client asyncdefsimple_client():"""使用 mcp 库的简化客户端""" url ="http://127.0.0.1:8000/sse"# 指向 SSE 挂载点try:asyncwith sse_client(url)as(read, write):asyncwith 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_date函数的参数。
完整的mcp_client.py客户端代码如下：
import asyncio from mcp import ClientSession from mcp.client.sse import sse_client asyncdefsimple_client():"""使用官方 mcp 库的简化客户端"""# 配置 SSE 服务器参数 url="http://127.0.0.1:8000/mcp"try:asyncwith sse_client(url)as(read, write):asyncwith 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_gStore_data 工具 call_result =await session.call_tool("get_gStore_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_date 函数被触发时打印的信息：
收到问题: 测试问题, 类型: test 结论
本文通过一个简单的示例，展示了如何使用 fastapi-mcp 和 mcp 库在 Python 中构建 MCP 服务端和客户端。这种模式非常适合于需要将现有 Web API（特别是 FastAPI）快速封装成可远程调用的“工具集”的场景，为服务间通信提供了一种灵活且强大的替代方案。通过 operation_id 和 tags 的智能映射，集成过程变得非常直观和高效。
