使用说明
MCP服务器实施指南
面向代理自动化解决方案架构师
本指南通过生产就绪的示例和架构模式全面了解MCP（模型上下文协议）服务器实现。
目录
架构概述
MCP堆栈
┌─────────────────────────────────────────────────────────────┐
│ Claude (LLM Host) │
│ - Receives user requests │
│ - Decides which tools to call │
│ - Processes tool responses │
└────────────────┬────────────────────────────────────────────┘
│ MCP Protocol (JSON-RPC 2.0)
│ Transport: stdio, SSE, HTTP
┌────────────────▼────────────────────────────────────────────┐
│ MCP Server (Your Code) │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ Tool Registration & Dispatch │ │
│ ├──────────────────────────────────────────────────────┤ │
│ │ Tool 1: search_knowledge_base() │ │
│ │ Tool 2: get_document() │ │
│ │ Tool 3: list_categories() │ │
│ └──────────────────────────────────────────────────────┘ │
└────────────────┬────────────────────────────────────────────┘
│ Your Business Logic
┌────────────────▼────────────────────────────────────────────┐
│ External Systems & Data │
│ - Databases (SQL, NoSQL, Vector) │
│ - APIs (REST, GraphQL, gRPC) │
│ - File Systems │
│ - Cloud Services (AWS, GCP, Azure) │
│ - Internal Services │
└─────────────────────────────────────────────────────────────┘
通信流
1. User Message → Claude
2. Claude analyzes message → Determines tool needed
3. Claude → MCP Server: call_tool(name="search_knowledge_base", args={...})
4. MCP Server → External System: Fetch data
5. External System → MCP Server: Return data
6. MCP Server → Claude: Format and return result
7. Claude → User: Synthesize response with tool data
核心概念
1. 工具 （主要机制）
工具是Claude可以调用的函数。它们是扩展克劳德能力的主要方式。
主要特征：
- 发现的：Claude可以列出所有可用的工具
- 自我描述：每个工具都有详细的描述和模式
- 打印的：输入/输出模式强制类型安全
- 无状态：每个调用都是独立的（尽管您可以实现state）
工具元数据：
@server.list_tools()
async def list_tools() -> list[Tool]:
return [
Tool(
name="tool_name",
description="What it does, when to use it, examples",
inputSchema=InputModel.model_json_schema()
)
]
2. 资源 （可选）
资源是克劳德可以读取的数据。与工具（Claude称之为工具）不同，资源是被动数据源。
使用案例：
- 配置文件
- 模板
- 静态文档
- 系统状态
3. 提示 （可选）
用户可以调用的预定义提示模板。
使用案例：
- 常见工作流程
- 标准化分析模式
- 模板化交互
4. 运输
Claude如何与您的服务器通信：
- 标准：标准输入/输出（最常见、最简单）
- 上海证券交易所：服务器发送事件（适用于web应用程序）
- 超文本传输协议：标准HTTP请求
实施演练
第一步：项目结构
my-mcp-server/
├── server.py # Main server code
├── requirements.txt # Dependencies
├── config.json # Configuration
├── README.md # Documentation
└── tests/ # Test suite
└── test_server.py
第二步：依赖关系
requirements.txt：
mcp>=0.9.0
pydantic>=2.0.0
aiohttp>=3.9.0 # If calling HTTP APIs
asyncpg>=0.29.0 # If using PostgreSQL
redis>=5.0.0 # If using Redis
安装：
pip install -r requirements.txt
步骤3：核心服务器结构
例子 knowledge_base_server.py 演示了基本组件：
A.进口和配置
from mcp.server import Server
from mcp.types import Tool, TextContent
from pydantic import BaseModel, Field, ConfigDict
B.输入模型（类型安全）
class SearchInput(BaseModel):
model_config = ConfigDict(extra='forbid') # Reject unknown fields
query: str = Field(
description="Search query",
min_length=1,
max_length=500,
examples=["authentication best practices"]
)
为什么是Pydantic？
- 自动验证
- JSON模式生成
- 类型检查
- 清除错误消息
C.服务器初始化
server = Server("my-server-name")
D.工具登记
@server.list_tools()
async def list_tools() -> list[Tool]:
return [Tool(...)]
E.工具实施
@server.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
if name == "my_tool":
input_data = MyInput(**arguments)
result = await do_something(input_data)
return [TextContent(type="text", text=result)]
F.主要入口点
async def main():
from mcp.server.stdio import stdio_server
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
server.create_initialization_options()
)
if __name__ == "__main__":
asyncio.run(main())
建筑师设计模式
模式1：搜索→ 提取模式
问题：大型数据集无法放在一个响应中
解决方案：两步模式
- 搜索工具返回ID和摘要
- Fetch工具按ID检索完整详细信息
# Step 1: Search (returns 10 results with IDs)
search_knowledge_base(query="kubernetes")
→ Returns: [doc-005: "Container Orchestration...", ...]
# Step 2: Fetch specific document
get_document(document_id="doc-005")
→ Returns: Full content of document
优点：
- 尊重上下文限制
- 允许细化
- 更高效的令牌使用
模式2：分页模式
问题：大型结果集
解决方案：返回带有标记的页面中的结果
class SearchInput(BaseModel):
query: str
page_size: int = Field(default=10, le=100)
page_token: Optional[str] = None
# Response includes next_page_token
{
"results": [...],
"next_page_token": "abc123"
}
模式3：格式灵活性模式
问题：不同的用例需要不同的格式
解决方案：支持JSON和Markdown
class SearchInput(BaseModel):
format: Literal["json", "markdown"] = "markdown"
detail_level: Literal["concise", "detailed"] = "concise"
何时使用每个：
- 标记语言：人类可读的最终答案
- JSON：机器可读、中间步骤、数据处理
模式4：复合操作模式
问题：常见工作流需要多个API调用
解决方案：创建组合操作的工作流工具
# Instead of:
# 1. check_availability()
# 2. create_event()
# 3. send_invites()
# Create:
async def schedule_meeting(params):
"""One tool that does all three"""
available_slots = await check_availability(...)
if available_slots:
event = await create_event(...)
await send_invites(...)
return event
模式5：错误恢复模式
问题：外部系统故障
解决方案：可操作的错误消息
except RateLimitError as e:
return [TextContent(
type="text",
text=f"Rate limit exceeded. Try again in {e.retry_after} seconds.\n\n"
f"Alternatively, reduce max_results or use more specific filters."
)]
生产注意事项
1.上下文窗口管理
字符限制：
CHARACTER_LIMIT = 25000 # ~6,250 tokens
def truncate_text(text: str, max_chars: int = CHARACTER_LIMIT) -> str:
if len(text) <= max_chars:
return text
return text[:max_chars - 50] + "\n\n[Content truncated]"
最佳实践：
- 默认为简洁的响应
- 提供detail_level参数
- 实现分页
- 优雅地截断
2.身份验证和安全
import os
from typing import Optional
class ServerConfig:
def __init__(self):
# Never hardcode secrets
self.api_key = os.environ.get("API_KEY")
self.database_url = os.environ.get("DATABASE_URL")
if not self.api_key:
raise ValueError("API_KEY environment variable required")
config = ServerConfig()
安全检查表：
- ✅ 使用环境变量作为机密
- ✅ 验证所有输入
- ✅ 对输出进行消毒（无敏感数据泄漏）
- ✅ 实施速率限制
- ✅ 对外部API使用HTTPS
- ✅ 记录安全事件（非敏感数据）
3.错误处理
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
try:
# Validate input
input_data = InputModel(**arguments)
# Execute operation
result = await perform_operation(input_data)
# Return result
return [TextContent(type="text", text=result)]
except ValidationError as e:
# Pydantic validation errors
return [TextContent(
type="text",
text=f"Invalid input: {e}\n\nPlease check parameter types and values."
)]
except ExternalAPIError as e:
# External service errors
return [TextContent(
type="text",
text=f"External service error: {e}\n\nThe service may be temporarily unavailable."
)]
except Exception as e:
# Unexpected errors
logger.exception(f"Unexpected error in {name}")
return [TextContent(
type="text",
text=f"An unexpected error occurred. Please try again or contact support."
)]
4.性能优化
# Connection pooling for databases
from asyncpg import create_pool
pool = await create_pool(
database_url,
min_size=5,
max_size=20,
command_timeout=10
)
# Caching for frequent requests
from functools import lru_cache
@lru_cache(maxsize=1000)
def get_cached_data(key: str):
return expensive_operation(key)
# Parallel requests
import asyncio
async def fetch_multiple(ids: list[str]):
tasks = [fetch_one(id) for id in ids]
return await asyncio.gather(*tasks)
5.测井和观测
import logging
import structlog
# Structured logging
logger = structlog.get_logger()
async def call_tool(name: str, arguments: Any):
logger.info(
"tool_called",
tool_name=name,
arguments=arguments
)
start_time = time.time()
try:
result = await execute_tool(name, arguments)
logger.info(
"tool_succeeded",
tool_name=name,
duration_ms=(time.time() - start_time) * 1000
)
return result
except Exception as e:
logger.error(
"tool_failed",
tool_name=name,
error=str(e),
duration_ms=(time.time() - start_time) * 1000
)
raise
测试和调试
单元测试工具
import pytest
from knowledge_base_server import search_documents
@pytest.mark.asyncio
async def test_search_documents():
results = await search_documents(
query="authentication",
max_results=5
)
assert len(results) > 0
assert all('id' in doc for doc in results)
assert results[0]['title'] # Has content
@pytest.mark.asyncio
async def test_search_no_results():
results = await search_documents(
query="nonexistent_term_xyz",
max_results=5
)
assert len(results) == 0
集成测试
import subprocess
import json
def test_server_startup():
"""Test that server starts without errors"""
process = subprocess.Popen(
['python', 'knowledge_base_server.py'],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
# Send initialize request
initialize_request = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {
"name": "test-client",
"version": "1.0.0"
}
}
}
process.stdin.write(json.dumps(initialize_request).encode() + b'\n')
process.stdin.flush()
# Check response (simplified)
# In real testing, parse JSON-RPC response
process.terminate()
process.wait(timeout=5)
调试提示
1.启用详细日志记录：
import logging
logging.basicConfig(level=logging.DEBUG)
2.在tmux中测试：
# Terminal 1: Run server
tmux new -s mcp
python knowledge_base_server.py
# Terminal 2: Send test requests
# (Use Claude Desktop or MCP inspector)
3.使用MCP检查器：
npx @modelcontextprotocol/inspector python knowledge_base_server.py
这将打开一个web UI，您可以在其中：
- 列出可用工具
- 使用自定义输入调用工具
- 查看JSON-RPC消息
- 调试响应
高级模式
模式1：数据库集成
import asyncpg
class DatabaseMCPServer:
def __init__(self):
self.pool = None
async def initialize(self):
self.pool = await asyncpg.create_pool(
'postgresql://user:pass@localhost/db'
)
async def query_database(self, sql: str, params: list):
async with self.pool.acquire() as conn:
return await conn.fetch(sql, *params)
# Tool implementation
async def search_users(query: str):
sql = """
SELECT id, name, email
FROM users
WHERE name ILIKE $1 OR email ILIKE $1
LIMIT 10
"""
results = await server.query_database(sql, [f'%{query}%'])
return format_results(results)
模式2:neneneba API与身份验证的集成
import aiohttp
from typing import Optional
class APIClient:
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.example.com"
self.session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
self.session = aiohttp.ClientSession(
headers={"Authorization": f"Bearer {self.api_key}"}
)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.session:
await self.session.close()
async def get(self, endpoint: str, params: dict = None):
async with self.session.get(
f"{self.base_url}/{endpoint}",
params=params,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
response.raise_for_status()
return await response.json()
# Usage in tool
async def fetch_user_data(user_id: str):
async with APIClient(config.api_key) as client:
user = await client.get(f"users/{user_id}")
return format_user(user)
模式3：文件系统访问
import os
import aiofiles
from pathlib import Path
async def search_files(directory: str, pattern: str):
"""Search files by content"""
results = []
base_path = Path(directory).resolve()
# Security: Prevent directory traversal
if not str(base_path).startswith(str(Path.home())):
raise ValueError("Access denied: outside home directory")
for file_path in base_path.rglob(pattern):
if file_path.is_file():
async with aiofiles.open(file_path, 'r') as f:
content = await f.read(1000) # First 1000 chars
results.append({
'path': str(file_path),
'preview': content
})
return results
模式4：矢量数据库集成
from typing import List
import numpy as np
class VectorSearchServer:
def __init__(self):
# In production, use Pinecone, Weaviate, Qdrant, etc.
self.embeddings = {} # document_id -> vector
self.documents = {} # document_id -> content
async def embed_text(self, text: str) -> np.ndarray:
"""Generate embedding (use OpenAI, Cohere, etc.)"""
# Simplified - use actual embedding model
return np.random.rand(768)
async def semantic_search(self, query: str, top_k: int = 5):
query_vector = await self.embed_text(query)
# Calculate cosine similarity
similarities = {}
for doc_id, doc_vector in self.embeddings.items():
similarity = np.dot(query_vector, doc_vector) / (
np.linalg.norm(query_vector) * np.linalg.norm(doc_vector)
)
similarities[doc_id] = similarity
# Return top-k results
top_docs = sorted(similarities.items(), key=lambda x: x[1], reverse=True)[:top_k]
return [
{
'id': doc_id,
'content': self.documents[doc_id],
'score': score
}
for doc_id, score in top_docs
]
模式5：工作流编排
async def create_and_deploy_service(params: dict):
"""
Multi-step workflow:
1. Create git repository
2. Initialize project structure
3. Deploy to cloud
4. Configure DNS
5. Set up monitoring
"""
results = {
'steps': [],
'status': 'in_progress'
}
try:
# Step 1: Create repo
repo_url = await create_github_repo(params['repo_name'])
results['steps'].append({'step': 'create_repo', 'status': 'success', 'url': repo_url})
# Step 2: Initialize project
await initialize_project_structure(repo_url, params['template'])
results['steps'].append({'step': 'initialize', 'status': 'success'})
# Step 3: Deploy
deployment = await deploy_to_cloud(repo_url, params['cloud_provider'])
results['steps'].append({'step': 'deploy', 'status': 'success', 'url': deployment['url']})
# Step 4: DNS
await configure_dns(params['domain'], deployment['ip'])
results['steps'].append({'step': 'dns', 'status': 'success'})
# Step 5: Monitoring
await setup_monitoring(deployment['url'])
results['steps'].append({'step': 'monitoring', 'status': 'success'})
results['status'] = 'completed'
except Exception as e:
results['status'] = 'failed'
results['error'] = str(e)
return results
配置和部署
Claude桌面配置
编辑 ~/Library/Application Support/Claude/claude_desktop_config.json （Mac）：
{
"mcpServers": {
"knowledge-base": {
"command": "python",
"args": ["/absolute/path/to/knowledge_base_server.py"],
"env": {
"PYTHONUNBUFFERED": "1",
"API_KEY": "your-api-key-here"
}
}
}
}
Linux： ~/.config/Claude/claude_desktop_config.json
窗户： %APPDATA%\Claude\claude_desktop_config.json
Docker部署
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY knowledge_base_server.py .
CMD ["python", "knowledge_base_server.py"]
Kubernetes部署
apiVersion: apps/v1
kind: Deployment
metadata:
name: mcp-server
spec:
replicas: 3
selector:
matchLabels:
app: mcp-server
template:
metadata:
labels:
app: mcp-server
spec:
containers:
- name: mcp-server
image: your-registry/mcp-server:latest
env:
- name: API_KEY
valueFrom:
secretKeyRef:
name: mcp-secrets
key: api-key
resources:
requests:
memory: "256Mi"
cpu: "200m"
limits:
memory: "512Mi"
cpu: "500m"
建筑师的现实世界用例
1.内部知识库
- 连接到Confluence、Notion或内部wiki
- 启用跨公司文档的自然语言搜索
- 减少搜索信息所花费的时间
2.数据库查询接口
- 允许自然语言数据库查询
- 实现只读视图以确保安全
- 按需生成报告和分析
3.DevOps自动化
- 检查部署状态
- 查看日志和指标
- 触发部署（具有审批工作流）
- 管理基础设施
4.客户支持集成
- 搜索支持票
- 获取客户信息
- 创建和更新工单
- 分析支持趋势
5.代码库管理
- 跨仓库搜索代码
- 审查PR和问题
- 生成文档
- 分析代码质量指标
建筑师的关键要点
-
为代理人而非人类设计
- 工具应该启用工作流，而不仅仅是包装API
- 针对有限的上下文进行优化
- 使错误可操作
-
从简单开始，迭代
- 从2-3个核心工具开始
- 真实用户测试（Claude）
- 根据实际需要添加工具
-
安全第一
- 验证所有输入
- 使用环境变量作为机密
- 实施正确的身份验证
- 记录安全事件
-
性能很重要
- 使用连接池
- 实现缓存
- 设置合理的超时
- 优雅地处理速率限制
-
可观察性至关重要
- 记录所有工具调用
- 跟踪成功/失败率
- 监控响应时间
- 异常警报
后续步骤
- ✅ 运行示例：测试知识库服务器
- 🔨 构建您的第一台服务器：从一个简单的用例开始
- 📚 阅读文档：研究MCP协议https://modelcontextprotocol.io
- 🧪 创建评估：使用现实场景测试您的服务器
- 🚀 部署：带监控转移到生产环境
其他资源
- MCP协议文件: https://modelcontextprotocol.io
- 开发包: https://github.com/modelcontextprotocol/python-sdk
- TypeScript SDK: https://github.com/modelcontextprotocol/typescript-sdk
- 示例服务器: https://github.com/modelcontextprotocol/servers
- 克劳德文件: https://docs.claude.com
问题或反馈？ 打开一个问题或为MCP社区做出贡献！
