MCP-Agent项目中的MCP服务器：架构原理与开发实践
2025-07-07 07:14:00作者：丁柯新Fawn
什么是MCP服务器？
在MCP-Agent项目中，MCP服务器（Model Context Protocol Servers）是支撑智能代理能力的核心组件。它们通过标准化的协议为代理提供各种功能扩展，可以理解为代理的"外接大脑"和"工具箱"。
MCP服务器的本质特征包括：
- 能力扩展器：为代理提供原生大模型不具备的专业功能
- 协议标准化：遵循统一的Model Context Protocol进行通信
- 独立进程：可以单独开发、部署和扩展
- 多形态支持：既可以是本地子进程，也可以是远程服务
MCP服务器的核心架构
通信传输层
MCP-Agent支持多种通信协议，适应不同场景需求：
-
STDIO（标准输入输出）
- 适用于本地开发调试
- 通过子进程方式启动服务器
- 配置示例：
filesystem: transport: "stdio" command: "npx" args: ["-y", "@modelcontextprotocol/server-filesystem"]
-
SSE（服务器发送事件）
- 适合实时数据流场景
- 支持长连接和事件推送
- 配置示例：
analytics: transport: "sse" url: "http://localhost:8000/sse"
-
WebSocket
- 全双工通信协议
- 适合需要双向交互的场景
- 配置示例：
chat: transport: "websocket" url: "ws://localhost:8001/ws"
-
Streamable HTTP
- 兼容性最好的传输方式
- 支持流式响应
- 配置示例：
api: transport: "streamable_http" url: "http://api.example.com/mcp"
服务器能力类型详解
1. 工具（Tools）
工具是代理可以调用的具体功能函数，相当于给代理"安装"的新技能。
开发示例：
@mcp.tool()
def calculate_metrics(data: str) -> dict:
"""分析数据并返回关键指标"""
# 实现数据分析逻辑
return {
"average": 42.5,
"max": 100,
"min": 10
}
2. 资源（Resources）
资源为代理提供数据访问能力，采用URI模式进行标识和访问。
开发示例：
@mcp.resource("db://users/{user_id}")
def get_user(user_id: str) -> dict:
"""根据ID获取用户信息"""
return {
"id": user_id,
"name": "张三",
"role": "developer"
}
3. 提示模板（Prompts）
可复用的提示模板，帮助代理生成更专业的输出。
开发示例：
@mcp.prompt()
def analysis_template(data: str) -> str:
"""数据分析提示模板"""
return f"""请分析以下数据：
{data}
要求：
1. 识别关键趋势
2. 标注异常值
3. 提出改进建议"""
服务器开发实战指南
使用FastMCP快速开发
FastMCP是推荐的MCP服务器开发框架，提供简洁的API：
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("数据分析服务器")
@mcp.tool()
def analyze_sales(period: str) -> dict:
"""销售数据分析"""
# 实现分析逻辑
return {"period": period, "growth": "15%"}
@mcp.resource("sales://{region}")
def get_regional_sales(region: str) -> str:
"""获取区域销售数据"""
return f"{region}区域Q3销售额：¥1,200万"
if __name__ == "__main__":
mcp.run()
生产环境最佳实践
-
输入验证
from pydantic import BaseModel class AnalysisRequest(BaseModel): data: str method: str = "standard" @validator('method') def validate_method(cls, v): if v not in ["standard", "advanced"]: raise ValueError("无效的分析方法") return v @mcp.tool() def perform_analysis(request: AnalysisRequest) -> str: """执行数据分析""" return f"使用{request.method}方法分析完成" -
异步处理
import aiohttp @mcp.tool() async def fetch_data(url: str) -> str: """异步获取数据""" async with aiohttp.ClientSession() as session: async with session.get(url) as resp: return await resp.text() -
错误处理
@mcp.tool() def safe_operation(param: str) -> str: """带有错误处理的操作""" try: # 可能失败的操作 return "操作成功" except Exception as e: return f"操作失败: {str(e)}"
高级功能：交互式确认
MCP服务器支持在执行过程中与用户进行交互确认：
from mcp.server.elicitation import AcceptedElicitation
class ConfirmParams(BaseModel):
confirm: bool
priority: str = "normal"
@mcp.tool()
async def process_order(order: str, ctx: Context) -> str:
"""处理订单前确认"""
result = await ctx.elicit(
message=f"确认处理订单：{order}",
schema=ConfirmParams
)
if isinstance(result, AcceptedElicitation) and result.data.confirm:
return f"订单已处理，优先级：{result.data.priority}"
return "订单取消"
生产环境部署建议
-
安全配置
production_db: transport: "streamable_http" url: "https://api.example.com/mcp" headers: Authorization: "Bearer ${SECURE_TOKEN}" auth: type: "bearer" token: "${DB_API_KEY}" -
性能优化
high_load_service: http_timeout_seconds: 30 read_timeout_seconds: 120 headers: Keep-Alive: "timeout=60" -
监控集成
import logging logger = logging.getLogger(__name__) @mcp.tool() def monitored_task(): """带监控的任务""" logger.info("任务开始") # 任务逻辑 logger.info("任务完成")
典型应用场景
-
数据访问层
- 封装数据库访问
- 提供统一的数据接口
- 实现数据缓存和优化
-
业务逻辑服务
- 订单处理
- 支付网关集成
- 库存管理
-
AI增强功能
- 专业领域模型
- 数据分析服务
- 图像/语音处理
通过MCP服务器的灵活组合，可以为MCP-Agent构建强大的能力矩阵，使其在保持核心架构简洁的同时，能够应对各种复杂业务场景的需求。
