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

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

# main.py
from 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="根据问题获取答案")
async def get_data(question: str, type: str):
    # 在实际应用中，这里会调用服务层处理业务逻辑
    # chat_service = get_chat_service()
    # answer = await chat_service.get_answer_mcp(question, type)
    print(f"收到问题：{question}, 类型：{type}")
    return f"这是关于 '{question}' 的答案。"
# ...

# 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)

# 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())

# 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}")

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())

python main.py

python test/mcp_client.py

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

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

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

引言

服务端实现 (`main.py`)

1. 定义 FastAPI 路由

2. 集成 `FastApiMCP`

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

1. 连接和初始化

2. 列出和调用工具

如何运行

1. 启动服务端

2. 运行客户端

预期输出

结论

更多推荐文章

相关免费在线工具

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

引言

服务端实现 (main.py)

1. 定义 FastAPI 路由

2. 集成 FastApiMCP

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

1. 连接和初始化

2. 列出和调用工具

如何运行

1. 启动服务端

2. 运行客户端

预期输出

结论

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具

服务端实现 (`main.py`)

2. 集成 `FastApiMCP`

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