主题
服务器概念
MCP 服务器是提供上下文、数据和功能给 AI 应用程序的核心组件。本页面详细介绍了服务器的核心概念和实现方式。
服务器功能
MCP 服务器可以提供三种主要类型的功能：
资源 (Resources)
资源是类似文件的数据，可以被客户端读取：
- API 响应数据
- 文件内容
- 数据库查询结果
- 实时数据流
特点：
- 只读访问
- 支持分页和流式传输
- 可以是静态或动态内容
工具 (Tools)
工具是可以被 LLM 调用的函数：
- 搜索引擎查询
- 计算器功能
- 文件操作
- API 调用
特点：
- 需要用户批准才能执行
- 支持参数验证
- 可以返回结构化结果
提示 (Prompts)
提示是帮助用户完成特定任务的预写模板：
- 代码生成模板
- 分析框架
- 写作指南
- 工作流程
特点：
- 可以包含参数化内容
- 支持多种格式
- 可以组合使用
协议消息
服务器通过以下协议消息与客户端交互：
列表操作
list_resources- 列出可用资源list_tools- 列出可用工具list_prompts- 列出可用提示
获取操作
read_resource- 读取特定资源get_prompt- 获取特定提示
调用操作
call_tool- 调用特定工具
订阅操作
subscribe- 订阅资源更新unsubscribe- 取消订阅
服务器生命周期
初始化阶段
- 启动：服务器进程启动
- 能力协商：与客户端交换能力信息
- 配置：设置运行参数和环境
运行阶段
- 监听请求：等待客户端消息
- 处理请求：执行相应的业务逻辑
- 返回响应：发送结果给客户端
关闭阶段
- 清理资源：释放占用的资源
- 保存状态：持久化必要的数据
- 优雅退出：正常关闭服务器
实现模式
基于 STDIO 的服务器
python
from mcp.server.fastmcp import FastMCP
# 创建服务器实例
mcp = FastMCP("my-server")
@mcp.resource("file://{path}")
async def read_file(path: str) -> str:
"""读取文件内容"""
with open(path, 'r') as f:
return f.read()
@mcp.tool()
async def calculate(expression: str) -> float:
"""计算数学表达式"""
return eval(expression) # 注意：实际使用中需要安全处理
if __name__ == "__main__":
mcp.run()基于 HTTP 的服务器
python
from fastapi import FastAPI
from mcp.server.fastmcp import FastMCP
app = FastAPI()
mcp = FastMCP("http-server")
@mcp.resource("api://data/{id}")
async def get_data(id: str) -> dict:
"""获取 API 数据"""
# 实现 API 数据获取逻辑
return {"id": id, "data": "..."}
# 集成到 FastAPI
app.mount("/mcp", mcp.create_app())错误处理
错误类型
- 协议错误：消息格式或协议违规
- 业务错误：功能执行失败
- 系统错误：资源不足或系统故障
错误响应
python
from mcp.types import McpError
@mcp.tool()
async def risky_operation(param: str) -> str:
try:
# 执行可能失败的操作
result = perform_operation(param)
return result
except ValueError as e:
raise McpError(
code="INVALID_PARAMS",
message=f"参数无效: {e}"
)
except Exception as e:
raise McpError(
code="INTERNAL_ERROR",
message="内部服务器错误"
)安全考虑
输入验证
- 验证所有输入参数
- 防止注入攻击
- 限制资源访问范围
权限控制
- 实现适当的访问控制
- 验证用户权限
- 记录敏感操作
资源限制
- 限制内存使用
- 控制执行时间
- 防止资源耗尽
性能优化
缓存策略
- 缓存频繁访问的数据
- 实现智能缓存失效
- 使用适当的缓存层次
异步处理
- 使用异步 I/O 操作
- 并发处理多个请求
- 避免阻塞操作
资源管理
- 及时释放资源
- 使用连接池
- 监控资源使用情况
日志记录
STDIO 服务器日志记录
python
import logging
# 配置日志输出到 stderr
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[logging.StreamHandler()] # 默认输出到 stderr
)
logger = logging.getLogger(__name__)
@mcp.tool()
async def logged_operation(param: str) -> str:
logger.info(f"执行操作，参数: {param}")
# 执行操作
result = "操作结果"
logger.info(f"操作完成，结果: {result}")
return resultHTTP 服务器日志记录
python
import logging
# HTTP 服务器可以使用标准输出
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)测试策略
单元测试
- 测试各个功能模块
- 模拟外部依赖
- 验证错误处理
集成测试
- 测试客户端-服务器交互
- 验证协议兼容性
- 测试端到端流程
性能测试
- 负载测试
- 压力测试
- 内存泄漏检测
部署考虑
环境配置
- 设置环境变量
- 配置依赖项
- 准备运行环境
监控和维护
- 实现健康检查
- 监控性能指标
- 设置告警机制
扩展性
- 支持水平扩展
- 实现负载均衡
- 考虑高可用性
下一步
了解了服务器概念后，您可以：