用 Python SDK 实现 MCP Server
Python 是实现 MCP Server 最流行的语言，官方 SDK 和 FastMCP 都提供了良好的支持。本节通过一个完整的实战案例，带你从零构建一个生产可用的 MCP Server。
环境准备
# 创建虚拟环境
python3 -m venv mcp-server-env
source mcp-server-env/bin/activate
# 安装依赖
pip install mcp # 官方 SDK
pip install fastmcp # 更简洁的封装层（推荐新手使用）
pip install httpx # 异步 HTTP 客户端
pip install pydantic # 数据验证（可选）
# 验证安装
python3 -c "import mcp; print(mcp.__version__)"
方式一：FastMCP（快速开始）
FastMCP 用装饰器语法大幅简化了 Server 的编写，5 分钟即可上手：
# fastmcp_example.py
from fastmcp import FastMCP
import httpx
mcp = FastMCP("天气助手")
@mcp.tool()
async def get_weather(city: str, unit: str = "celsius") -> str:
"""
获取指定城市的天气信息。
适用于用户询问天气、出行建议等场景。
Args:
city: 城市名称，如"北京"、"上海"
unit: 温度单位，celsius（摄氏）或 fahrenheit（华氏），默认 celsius
"""
# 实际应用中替换为真实的天气 API
async with httpx.AsyncClient() as client:
response = await client.get(
"https://api.weather.example.com/current",
params={"city": city, "unit": unit}
)
data = response.json()
return f"{city}当前天气：{data['description']}，温度 {data['temp']}°{'C' if unit == 'celsius' else 'F'}，湿度 {data['humidity']}%"
@mcp.tool()
async def get_forecast(city: str, days: int = 3) -> str:
"""
获取城市未来天气预报。
Args:
city: 城市名称
days: 预报天数，1-7，默认3
"""
if not 1 <= days <= 7:
return "错误：days 参数必须在 1-7 之间"
# 构建预报内容...
return f"{city} 未来{days}天预报：..."
if __name__ == "__main__":
mcp.run()
配置到 Claude Desktop：
{
"mcpServers": {
"weather": {
"command": "/path/to/mcp-server-env/bin/python3",
"args": ["/path/to/fastmcp_example.py"]
}
}
}
方式二：官方 SDK（完整控制）
对于需要更精细控制的场景，使用官方 SDK：
完整实战：内部知识库检索 MCP
# knowledge_base_server.py
"""
内部知识库检索 MCP Server
功能：搜索 Markdown 文档库，返回相关内容
"""
import asyncio
import json
import os
from pathlib import Path
from typing import Optional
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
# === 初始化 ===
app = Server("knowledge-base")
KB_DIR = Path(os.environ.get("KB_DIR", os.path.expanduser("~/knowledge-base")))
# === 工具实现 ===
def search_documents(query: str, max_results: int = 5) -> list[dict]:
"""简单的全文搜索（生产环境可替换为 embedding 检索）"""
results = []
query_lower = query.lower()
for md_file in KB_DIR.rglob("*.md"):
try:
content = md_file.read_text(encoding="utf-8")
if query_lower in content.lower():
# 找到匹配行及上下文
lines = content.split("\n")
matching_lines = [
(i, line) for i, line in enumerate(lines)
if query_lower in line.lower()
]
if matching_lines:
# 取第一个匹配的上下文（前后各2行）
idx = matching_lines[0][0]
context_start = max(0, idx - 2)
context_end = min(len(lines), idx + 3)
context = "\n".join(lines[context_start:context_end])
results.append({
"file": str(md_file.relative_to(KB_DIR)),
"title": lines[0].lstrip("#").strip() if lines else md_file.stem,
"context": context,
"matches": len(matching_lines)
})
except Exception:
continue
# 按匹配数量排序
results.sort(key=lambda x: x["matches"], reverse=True)
return results[:max_results]
def get_document(file_path: str) -> Optional[str]:
"""读取完整文档"""
target = KB_DIR / file_path
# 安全检查：确保文件在 KB_DIR 内
try:
target.resolve().relative_to(KB_DIR.resolve())
except ValueError:
return None
if not target.exists() or not target.is_file():
return None
return target.read_text(encoding="utf-8")
# === MCP 工具注册 ===
@app.list_tools()
async def list_tools() -> list[types.Tool]:
return [
types.Tool(
name="search_knowledge_base",
description="""
搜索内部知识库文档。
适用场景：用户询问公司政策、产品文档、操作手册等内部知识。
返回：最相关的文档片段及来源文件名。
注意：只能搜索已同步到本地知识库的文档。
""",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词或问题"
},
"max_results": {
"type": "integer",
"default": 5,
"minimum": 1,
"maximum": 20,
"description": "返回结果数量，默认5"
}
},
"required": ["query"]
}
),
types.Tool(
name="get_full_document",
description="""
获取知识库文档的完整内容。
通常在 search_knowledge_base 之后调用，获取某篇文档的完整内容。
参数 file_path 来自 search_knowledge_base 返回的 file 字段。
""",
inputSchema={
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "文档相对路径，如 'policies/leave.md'"
}
},
"required": ["file_path"]
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
if name == "search_knowledge_base":
query = arguments["query"]
max_results = arguments.get("max_results", 5)
results = search_documents(query, max_results)
if not results:
return [types.TextContent(
type="text",
text=f"在知识库中未找到与'{query}'相关的文档。"
)]
# 格式化结果
output_lines = [f"找到 {len(results)} 个相关文档：\n"]
for i, result in enumerate(results, 1):
output_lines.append(f"**{i}. {result['title']}**")
output_lines.append(f"文件：`{result['file']}`（{result['matches']} 处匹配）")
output_lines.append(f"相关内容：\n```\n{result['context']}\n```\n")
return [types.TextContent(type="text", text="\n".join(output_lines))]
elif name == "get_full_document":
file_path = arguments["file_path"]
content = get_document(file_path)
if content is None:
return [types.TextContent(
type="text",
text=f"错误：文件 '{file_path}' 不存在或不在知识库目录内。"
)]
return [types.TextContent(type="text", text=content)]
else:
raise ValueError(f"未知工具: {name}")
# === 启动入口 ===
async def main():
if not KB_DIR.exists():
print(f"错误：知识库目录 {KB_DIR} 不存在", flush=True)
import sys
sys.exit(1)
async with stdio_server() as (read_stream, write_stream):
await app.run(
read_stream,
write_stream,
app.create_initialization_options()
)
if __name__ == "__main__":
asyncio.run(main())
配置：
{
"mcpServers": {
"knowledge-base": {
"command": "/path/to/venv/bin/python3",
"args": ["/path/to/knowledge_base_server.py"],
"env": {
"KB_DIR": "/Users/yourname/company-docs"
}
}
}
}
测试你的 Server
在接入 Claude Desktop 之前，先用命令行测试：
# 测试工具列表
echo '{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}' | \
python3 knowledge_base_server.py
# 测试工具调用
echo '{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"search_knowledge_base","arguments":{"query":"年假政策"}}}' | \
python3 knowledge_base_server.py
或者使用 MCP Inspector：
npx @modelcontextprotocol/inspector python3 /path/to/knowledge_base_server.py
本节执行清单
- [ ] 安装 Python MCP SDK 和 FastMCP
- [ ] 从 FastMCP 开始：用装饰器定义第一个工具
- [ ] 完整 SDK：使用
@app.list_tools()和@app.call_tool()结构 - [ ] 测试：先用命令行 JSON-RPC，再用 MCP Inspector，最后接入客户端