## 快速参考 **MCP 服务器开发框架** **它的作用**：使用 FastMCP 框架构建、测试和部署自定义 MCP（模型上下文协议）服务器的综合指南，用于向 Claude 和其他 AI 模型公开工具、资源和提示。 **核心能力**： * ✅ 使用类型安全装饰器进行 FastMCP 服务器开发 * ✅ 工具/资源/提示架构模式 * ✅ Pydantic 验证和错误处理 * ✅ OAuth2 和 API 密钥身份验证模式 * ✅ 性能监控和健康检查 * ✅ Docker 和 Kubernetes 部署 * ✅ 测试策略和验证 * ✅ 生产级模式（缓存、断路器、速率限制） **何时使用**： * 为内部工具构建自定义 MCP 服务器 * 通过MCP协议公开现有服务 * 实施企业身份验证模式 * 在生产环境中部署MCP服务器 * 优化服务器性能和可靠性 * 测试MCP服务器实现 ** ** ## 实施指南 ### FastMCP 入门 **安装**： pip 安装 fastmcp **最低服务器**：从 fastmcp 导入 FastMCP 服务器= FastMCP（“我的服务器”） @服务器.工具() def hello_world(名称: str) -> str: ”““向某人打招呼。”“” return f“你好，{name}！” 如果 __name__ == "__main__": 服务器.run() **核心概念**： 1. **工具**：Claude 可以使用经过验证的参数调用的函数 2. **资源**：用于公开信息的基于 URI 的数据端点 3. **提示**：可重复使用的对话模板和系统提示详细指南：[getting-started.md]() ** ** ### MCP 服务器架构 **三成分模式**： ┌──────────────────────────────────────────┐ │ MCP 服务器（FastMCP） │ ├──────────────────────────────────────────┤ │ 工具（功能） │ │ • @server.tool() 装饰器 │ │ • Pydantic 验证 │ │ • 工作流程优化的命名 │ │ │ │ 资源（数据端点） │ │ • @server.resource("uri://...") 装饰 │ │ • 流媒体支持 │ │ • 基于权限的访问 │ │ │ │ 提示（模板） │ │ • @server.prompt("name") 装饰器 │ │ • 参数注入 │ │ • 多轮工作流程 │ └──────────────────────────────────────────┘ ↓ MCP 协议 (JSON-RPC 2.0) ↓ ┌──────────────────────────────────────────┐ │ 克劳德 / 法学硕士客户 │ └──────────────────────────────────────────┘ **设计模式**：[server-design.md]() ** ** ### 生产就绪服务器示例从 fastmcp 导入 FastMCP from pydantic import 字段 from 输入 import Literal，可选服务器 = FastMCP("企业数据库服务器") @服务器.工具() def 搜索记录( 查询：str，表：文字[“用户”，“产品”，“订单”]，限制: int = Field(默认=10, ge=1, le=100), 过滤器：可选[dict] =无 ) -> 字典： ”“” 使用分页搜索数据库记录。参数：查询：搜索查询字符串 table：要搜索的表限制：最大结果 (1-100) 过滤器：可选的过滤条件返回：包含结果和元数据的字典 ”“” 如果不是查询或者不是query.strip()： raise ValueError("查询不能为空") 结果=execute_search(查询、表、限制、过滤器) 返回{ “状态”：“成功”， “计数”：len（结果）， “结果”：结果， “total_available”：get_total_count（查询） } @server.resource("db://{table}/{id}") def get_record(表: str, id: str) -> 字典: """通过ID获取记录。""" 记录 = fetch_record(表, id) 如果不记录： raise ValueError(f"未找到记录：{table}/{id}") 返回记录如果 __name__ == "__main__": 服务器.run() **实施指南**：[implementation.md]() ** ** ### 身份验证模式 **OAuth2（用户验证）**：从 fastmcp.auth 导入 OAuth2Provider oauth = OAuth2Provider( authorize_url =“https://auth.company.com/authorize”， token_url =“https://auth.company.com/token”，范围=[“读：数据”，“写：数据”] ） @server.auth(oauth) @服务器.工具() def protected_action(user_id: str) -> 字典: """需要 OAuth 令牌。""" 返回执行操作（用户id） **API 密钥（服务到服务）**：从 fastmcp.auth 导入 APIKeyAuth api_auth = APIKeyAuth(标题=“X-API-Key”) @server.auth(api_auth) @server.resource("安全://{resource_id}") def secure_resource(resource_id: str) -> str: """需要 API 密钥。""" 返回 fetch_data(resource_id) **详细模式**：[auth-patterns.md]() ** ** ### 测试 MCP 服务器 **单元测试**：导入pytest 从 fastmcp 导入 FastMCP @pytest.fixture 默认服务器（）： s = FastMCP(“测试服务器”) @s.tool() def add(a: int, b: int) -> int: 返回 a + b 返回 def test_add_tool(服务器): 结果 = server.invoke_tool("add", {"a": 2, "b": 3}) 断言结果 == 5 def test_invalid_params（服务器）：与 pytest.raises(ValueError): server.invoke_tool("add", {"a": "not-a-number", "b": 3}) **测试指南** : [testing.md]() ** ** ### 部署 **码头工人**：来自 python:3.11-slim 工作目录/应用程序复制requirements.txt。运行 pip install -rrequirements.txt 复制 server.py 。曝光 8000 CMD [“python”，“server.py”] **库伯内特**： api版本：apps/v1 种类：部署元数据：名称： mcp 服务器规格：副本：3 选择器：匹配标签：应用程序：mcp-服务器模板：元数据：标签：应用程序：mcp-服务器规格：容器： - 名称：mcp-服务器图片：mcp-服务器：最新端口： - 集装箱端口：8000 活性探针： http获取：路径：/健康端口：8000 初始延迟秒数：10 周期秒：10 **部署指南**：[deployment.md]() ** ** ## 高级模式 ### 工具设计最佳实践 ✅ **做**： * 为单个有意义的工作流程步骤设计工具 * 使用 Pydantic Field 约束验证所有输入 * 提供清晰、可操作的错误消息 * 对大型结果集使用分页 * 包括全面的文档字符串 ❌ **不要**： * 将多种职责混合在一个工具中 * 返回无限的结果集 * 跳过输入验证 * 暴露敏感数据 * 忽略错误处理 **工具设计指南** : [tool-design.md]() ** ** ### 性能优化 **缓存策略**：从 functools 导入包装从日期时间导入日期时间，时间增量 MCPCache 类： def __init__(自身，ttl_秒=300): 自我缓存 = {} self.ttl = ttl_秒 def 缓存(self, func): @wraps(函数) def 包装器（*args，**kwargs）： key = str((func.__name__, args, kwargs)) 如果在 self.cache 中键入：值，时间戳 = self.cache[key] if (datetime.now() - 时间戳).total_seconds() < self.ttl: 返回值结果 = func(*args, **kwargs) self.cache[key] = (结果, datetime.now()) 返回结果返回包装器缓存 = MCPCache(ttl_秒=600) @cache.cached def昂贵_操作（参数：str）->字典：返回 fetch_and_process(参数) **高级模式**： * [缓存]() * [断路器]() * [速率限制]() ** ** ### 监控和可观察性 **健康检查**： @server.resource(“健康状况://状态”) def health_check() -> 字典： """服务器健康状态。""" 返回{ “状态”：“健康”， “版本”：“1.0.0”， “uptime_seconds”：get_uptime（）， “活动连接”：get_connection_count() } **指标和日志记录**：导入日志记录导入时间记录器=logging.getLogger(__name__) @服务器.工具() def Monitored_Operation(params: dict) -> dict: 开始 = 时间.time() 尝试：结果=执行操作（参数）持续时间 = time.time() - 开始 logger.info(f"操作在 {duration:.2f}s 内完成") 返回结果除了异常 e： logger.error(f"操作失败：{str(e)}") 提高 **监控指南** : [monitoring.md]() ** ** ## 配合得很好 * `moai-context7-integration` \- API 模式的文档访问 * `moai-cc-configuration` \- MCP 服务器配置管理 * `moai-essentials-debug` \- 服务器调试和故障排除 * `moai-domain-backend` \- 后端服务架构 * `moai-domain-cloud` \- 云部署模式 * `moai-quality-security` \- 安全验证和 OWASP 合规性 ** ** ## 核心概念 1. **FastMCP Framework**：用于快速MCP服务器开发的Python库 2. **类型安全**：Pydantic 模型确保 Claude 了解参数约束 3. **工作流程设计**：用于单个有意义任务的工具，而不是细粒度的 API 4. **身份验证策略**：用于用户应用程序的 OAuth2，用于服务到服务的 API 密钥 5. **生产准备**：监控、运行状况检查、错误处理、缓存 6. **测试**：部署前的全面测试覆盖率 ** ** ## 模块导航 **开始**： * [getting-started.md]() \- FastMCP 基础知识 **核心开发**： * [server-design.md]() \- 架构模式 * [implementation.md]() \- 真实示例 * [testing.md]() \- 测试策略 **部署和操作**： * [deployment.md]() \- Docker/Kubernetes **高级模式**（模块/开发/模式/）： * [tool-design.md]() \- 工具最佳实践 * [resource-design.md]() \- 资源模式 * [auth-patterns.md]() \- 身份验证 * [monitoring.md]() \- 可观察性 * [caching.md]() \- 缓存策略 * [Circuit-breaker.md]() \- 弹性 * [rate-limiting.md]() \- 速率控制 ** ** ## 变更日志 * **v3.0.0** (2025-11-27)：从服务器集成到以服务器开发为中心的全面重构，开发模式模块化 * **v2.1.0** (2025-11-22)：模块化结构 - 重构SKILL.md，添加reference.md和examples.md * **v2.0.0** (2025-11-22)：MCP 1.0+ 协议完整规范更新 * **v1.0.0** (2025-11-21): 初始 MCP 集成技巧 ** ** **状态**：生产就绪|有关详细模式，请参阅“模块/开发/” |最后更新：2025-11-27

Moai 连接器 MCP

包含技能