大家好，我是讯享网，很高兴认识大家。这里提供最前沿的Ai技术和互联网信息。

# 从零构建Python版MCP服务器：以Claude Desktop为例的实战指南

在AI工具生态爆炸式增长的今天，Model Context Protocol（MCP）正在成为连接AI模型与外部服务的"万能适配器"。想象一下，你的AI助手不仅能回答问题，还能直接操作Notion数据库、触发GitHub Actions甚至控制智能家居——这正是MCP赋予我们的能力。本文将带你用Python打造一个专为Claude Desktop设计的MCP服务器，并通过GitHub Actions实现自动化部署，让你亲手搭建这个"AI与真实世界的桥梁"。

1. 环境准备与MCP基础架构

1.1 Python开发环境配置

构建MCP服务器需要以下基础环境：

# 创建虚拟环境（推荐使用Python 3.9+） python -m venv mcp-env source mcp-env/bin/activate # Linux/Mac mcp-envScriptsactivate # Windows # 安装核心依赖 pip install fastapi uvicorn python-dotenv 

关键组件对比表：

组件	作用	替代方案
FastAPI	提供HTTP接口	Flask, Django
Uvicorn	ASGI服务器	Hypercorn, Daphne
python-dotenv	环境变量管理	直接使用os.environ

> 提示：建议使用PyCharm或VS Code作为IDE，它们对API开发有更好的支持

1.2 MCP服务器三大核心模块

一个完整的MCP服务器需要实现三个基本功能模块：

1. 工具模块：处理外部API调用
   • 示例：Notion API操作、GitHub仓库管理
2. 资源模块：提供数据访问接口
   • 示例：数据库连接、文件系统访问
3. 提示模块：管理预定义工作流
   • 示例：客户支持话术模板、代码生成规则

# 基础MCP服务器结构示例 from fastapi import FastAPI app = FastAPI() @app.get("/mcp/tools") async def list_tools(): return {"tools": ["notion", "github"]} @app.post("/mcp/tools/notion") async def use_notion_tool(params: dict): # 实现Notion API调用逻辑 return {"status": "success"} 

2. Claude Desktop集成实战

2.1 建立双向通信通道

Claude Desktop作为MCP主机，需要通过WebSocket与我们的服务器保持长连接：

# websocket_endpoint.py from fastapi import WebSocket @app.websocket("/mcp/ws") async def websocket_endpoint(websocket: WebSocket): await websocket.accept() while True: data = await websocket.receive_json() # 处理来自Claude的请求 response = process_mcp_request(data) await websocket.send_json(response) 

通信协议要点：

• 使用JSON格式传输数据
• 每条消息必须包含message_id用于请求响应匹配
• 错误响应需遵循{"error": {"code": string, "message": string}}格式

2.2 实现Notion集成案例

让我们以实现Notion集成为例，展示完整工具开发流程：

1. 首先在Notion开发者平台创建integration并获取API密钥
2. 实现数据库查询工具：

# notion_integration.py import httpx from pydantic import BaseModel class NotionQuery(BaseModel): database_id: str filter: dict = None @app.post("/mcp/tools/notion/query") async def query_notion_database(query: NotionQuery): headers = { "Authorization": f"Bearer {NOTION_API_KEY}", "Notion-Version": "2022-06-28" } async with httpx.AsyncClient() as client: response = await client.post( f"https://api.notion.com/v1/databases/{query.database_id}/query", json={"filter": query.filter} if query.filter else {}, headers=headers ) return response.json() 

> 注意：务必在.env文件中存储NOTION_API_KEY，不要硬编码在代码中

3. 安全加固策略

3.1 身份验证与授权

MCP服务器必须实现严格的身份验证：

# auth_middleware.py from fastapi import Request, HTTPException from fastapi.security import APIKeyHeader api_key_header = APIKeyHeader(name="X-MCP-Key") async def verify_api_key(request: Request): api_key = request.headers.get("x-mcp-key") if api_key != os.getenv("MCP_API_KEY"): raise HTTPException(status_code=403, detail="Invalid API key") return True 

安全措施对照表：

风险类型	防护措施	实现方式
未授权访问	API密钥验证	中间件检查
数据泄露	传输加密	HTTPS/WSS
注入攻击	输入验证	Pydantic模型
暴力激活成功教程	速率限制	FastAPI限流

3.2 沙箱环境隔离

对于高风险操作，应使用容器隔离：

# 使用Docker运行不可信工具 docker run --rm -v $(pwd)/sandbox:/sandbox python:3.9-slim python /sandbox/untrusted_script.py 

对应的Python控制代码：

# sandbox.py import docker from docker.errors import DockerException client = docker.from_env() def run_in_sandbox(script_path: str): try: container = client.containers.run( "python:3.9-slim", f"python /sandbox/{os.path.basename(script_path)}", volumes={os.path.abspath(script_path): { 'bind': f'/sandbox/{os.path.basename(script_path)}', 'mode': 'ro' }}, network_mode="none", # 禁用网络 mem_limit="100m", # 内存限制 remove=True # 运行后自动删除 ) return container.decode('utf-8') except DockerException as e: return str(e) 

4. 自动化部署与CI/CD

4.1 GitHub Actions工作流配置

创建.github/workflows/deploy.yml实现自动部署：

name: Deploy MCP Server on: push: branches: [ main ] pull_request: branches: [ main ] jobs: deploy: runs-on: ubuntu-latest steps: - uses: actions/checkout@v3 - name: Set up Python uses: actions/setup-python@v4 with: python-version: '3.9' - name: Install dependencies run: | python -m pip install --upgrade pip pip install -r requirements.txt - name: Run tests run: | pytest - name: Deploy to AWS if: github.ref == 'refs/heads/main' uses: aws-actions/configure-aws-credentials@v2 with: aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }} aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }} aws-region: us-east-1 - name: Deploy with Terraform if: github.ref == 'refs/heads/main' run: | cd infra terraform init terraform apply -auto-approve 

4.2 监控与日志收集

使用Prometheus和Grafana搭建监控看板：

# monitoring.py from prometheus_fastapi_instrumentator import Instrumentator Instrumentator().instrument(app).expose(app) 

对应的Grafana仪表板配置要点：

• 请求成功率（HTTP状态码分布）
• 响应时间百分位（P50/P95/P99）
• 并发连接数
• 工具调用频率

日志收集架构：

1. 使用Fluentd收集容器日志
2. 通过Elasticsearch存储日志数据
3. 在Kibana中实现可视化查询

5. 高级功能扩展

5.1 动态工具注册系统

实现无需重启的热加载工具：

# dynamic_tools.py from fastapi import APIRouter class ToolRegistry: def __init__(self): self._tools = {} self._router = APIRouter() def register_tool(self, name: str, endpoint: str, handler): self._tools[name] = {"endpoint": endpoint, "handler": handler} self._router.add_api_route( f"/tools/{endpoint}", handler, methods=["POST"] ) def get_router(self): return self._router tool_registry = ToolRegistry() # 使用示例 @tool_registry.register_tool("weather", "get-weather") async def get_weather(location: str): # 调用天气API return {"temperature": 25, "condition": "sunny"} app.include_router(tool_registry.get_router()) 

5.2 性能优化技巧

对于高频率调用的工具，实现缓存层：

# cache_decorator.py from functools import wraps from datetime import timedelta from cachetools import TTLCache cache = TTLCache(maxsize=1000, ttl=timedelta(minutes=5)) def cached_tool(func): @wraps(func) async def wrapper(*args, kwargs): cache_key = f"{func.__name__}:{str(args)}:{str(kwargs)}" if cache_key in cache: return cache[cache_key] result = await func(*args, kwargs) cache[cache_key] = result return result return wrapper # 使用示例 @tool_registry.register_tool("stock", "get-stock-price") @cached_tool async def get_stock_price(symbol: str): # 调用股票API return {"price": 150.75} 

在实际项目中，我们还需要考虑工具版本兼容性、灰度发布策略以及灾备方案。例如，可以为每个工具添加版本标识，允许新旧版本并行运行一段时间：

# versioned_tool.py @app.post("/mcp/tools/v1/notion/query") async def query_notion_v1(query: NotionQueryV1): # 旧版实现 ... @app.post("/mcp/tools/v2/notion/query") async def query_notion_v2(query: NotionQueryV2): # 新版实现 ... 

通过GitHub Actions的部署策略，我们可以轻松实现蓝绿部署或金丝雀发布，确保服务更新的平滑过渡。