Awesome-MCP-Scaffold

# OutputSchema 使用指南 ## 📋 概述 本指南介绍如何在 Awesome-MCP-Scaffold 中使用 MCP SDK v1.11.0 的 `outputSchema` 功能，为工具提供结构化输出支持。 ## 🎯 什么是 OutputSchema `outputSchema` 是 MCP 协议的一个新特性，允许工具定义其返回值的结构化模式。这提供了： - **类型安全**：确保工具返回预期格式的数据 - **自动验证**：MCP SDK 自动验证返回值符合定义的模式 - **更好的集成**：客户端可以可靠地解析和使用工具返回的数据 - **向后兼容**：同时提供结构化和文本格式的输出 ## 🔧 快速开始 ### 1. 定义响应模型 ```python # server/models/responses.py from pydantic import BaseModel, Field from datetime import datetime class CalculationResult(BaseModel): """计算结果模型""" success: bool = Field(description="操作是否成功") timestamp: datetime = Field(default_factory=datetime.now, description="操作时间") result: float = Field(description="计算结果") expression: str = Field(description="计算表达式") operation: str = Field(description="操作类型") ``` ### 2. 创建结构化工具 ```python # server/tools/calculator.py from ..models.responses import CalculationResult @mcp.tool(title="Add Numbers", description="Add two numbers together") def add(a: float, b: float) -> CalculationResult: """Add two numbers and return structured result.""" result = a + b return CalculationResult( success=True, result=result, expression=f"{a} + {b}", operation="addition" ) ``` ### 3. 自动 Schema 生成 MCP SDK 会自动从返回类型注解生成 `outputSchema`： ```json { "name": "add", "title": "Add Numbers", "description": "Add two numbers together", "outputSchema": { "type": "object", "properties": { "success": {"type": "boolean", "description": "操作是否成功"}, "timestamp": {"type": "string", "format": "date-time", "description": "操作时间"}, "result": {"type": "number", "description": "计算结果"}, "expression": {"type": "string", "description": "计算表达式"}, "operation": {"type": "string", "description": "操作类型"} }, "required": ["success", "timestamp", "result", "expression", "operation"] } } ``` ## 📚 内置响应模型 脚手架提供了一套标准的响应模型： ### 基础模型 ```python from server.models.responses import BaseToolResponse class BaseToolResponse(BaseModel): """所有工具响应的基类""" success: bool = Field(description="操作是否成功") timestamp: datetime = Field(default_factory=datetime.now, description="操作时间") ``` ### 计算器模型 ```python from server.models.responses import CalculationResult, BMIResult, StatisticsResult # 基础计算结果 CalculationResult( success=True, result=42.0, expression="6 * 7", operation="multiplication" ) # BMI 计算结果 BMIResult( success=True, bmi=22.86, category="Normal weight", weight_kg=70.0, height_m=1.75 ) # 统计分析结果 StatisticsResult( success=True, average=15.5, count=4, sum=62.0, minimum=10.0, maximum=20.0 ) ``` ### 文本处理模型 ```python from server.models.responses import TextAnalysisResult, CaseConversionResult, ExtractionResult # 文本分析结果 TextAnalysisResult( success=True, words=50, characters=300, characters_no_spaces=250, lines=5 ) # 大小写转换结果 CaseConversionResult( success=True, original="Hello World", converted="HELLO WORLD", case_type="upper" ) # 文本提取结果 ExtractionResult( success=True, matches=["user@example.com", "admin@domain.org"], count=2, pattern="email" ) ``` ### 文件操作模型 ```python from server.models.responses import FileOperationResult, FileInfoResult, DirectoryResult # 文件操作结果 FileOperationResult( success=True, message="File written successfully", file_path="data/example.txt", size_bytes=1024, operation="write" ) # 文件信息结果 FileInfoResult( success=True, path="data/file.txt", name="file.txt", type="file", size_bytes=2048, created=1640995200.0, modified=1640995300.0, exists=True ) # 目录列表结果 DirectoryResult( success=True, path="data/", files=["file1.txt", "file2.json"], directories=["subdir1", "subdir2"], total_files=2, total_directories=2 ) ``` ## 🛠️ 自定义响应模型 ### 1. 创建新模型 ```python # server/models/responses.py class WeatherResult(BaseToolResponse): """天气查询结果""" temperature: float = Field(description="温度（摄氏度）") humidity: float = Field(description="湿度（百分比）", ge=0, le=100) condition: str = Field(description="天气状况") location: str = Field(description="查询位置") wind_speed: Optional[float] = Field(None, description="风速（km/h）") ``` ### 2. 使用自定义模型 ```python @mcp.tool(title="Get Weather", description="Get current weather information") def get_weather(city: str) -> WeatherResult: """Get weather information for a city.""" # 实际的天气 API 调用逻辑 weather_data = fetch_weather_api(city) return WeatherResult( success=True, temperature=weather_data["temp"], humidity=weather_data["humidity"], condition=weather_data["condition"], location=city, wind_speed=weather_data.get("wind_speed") ) ``` ## 🔄 迁移现有工具 ### 原始工具（返回 dict） ```python @mcp.tool(title="Old Tool", description="Legacy tool") def old_tool(param: str) -> dict: return { "result": "some value", "status": "ok" } ``` ### 迁移后工具（结构化输出） ```python from ..models.responses import SimpleResult @mcp.tool(title="New Tool", description="Modernized tool") def new_tool(param: str) -> SimpleResult: return SimpleResult( success=True, result="some value", details="Operation completed successfully" ) ``` ## 🧪 测试结构化输出 ### 1. 验证 Schema 生成 ```python async def test_tool_output_schema(): tools = await client.list_tools() tool = next(t for t in tools.tools if t.name == "add") assert tool.outputSchema is not None assert tool.outputSchema["type"] == "object" assert "result" in tool.outputSchema["properties"] ``` ### 2. 验证结构化输出 ```python async def test_structured_output(): result = await client.call_tool("add", {"a": 5, "b": 3}) # 验证结构化内容存在 if hasattr(result, 'structuredContent'): assert result.structuredContent["success"] is True assert result.structuredContent["result"] == 8.0 ``` ### 3. 验证向后兼容性 ```python async def test_backward_compatibility(): result = await client.call_tool("add", {"a": 10, "b": 20}) # 验证传统 content 仍然存在 assert hasattr(result, 'content') assert len(result.content) > 0 ``` ## 📊 响应格式示例 ### MCP 协议响应 ```json { "jsonrpc": "2.0", "id": 1, "result": { "content": [ { "type": "text", "text": "{\"success\": true, \"timestamp\": \"2025-01-01T12:00:00\", \"result\": 8.0, \"expression\": \"5 + 3\", \"operation\": \"addition\"}" } ], "structuredContent": { "success": true, "timestamp": "2025-01-01T12:00:00", "result": 8.0, "expression": "5 + 3", "operation": "addition" } } } ``` ### 客户端使用 ```python # 使用结构化内容（推荐） if hasattr(result, 'structuredContent'): data = result.structuredContent print(f"Result: {data['result']}") print(f"Operation: {data['operation']}") # 使用文本内容（向后兼容） else: text = result.content[0].text data = json.loads(text) print(f"Result: {data['result']}") ``` ## 🚨 最佳实践 ### 1. 模型设计 - **继承基类**：所有响应模型都应继承 `BaseToolResponse` - **描述完整**：使用 `Field(description=...)` 提供清晰的字段说明 - **类型安全**：使用适当的类型约束（如 `ge=0` 表示大于等于0） - **可选字段**：使用 `Optional` 和默认值处理可选数据 ### 2. 错误处理 ```python @mcp.tool(title="Safe Tool", description="Tool with error handling") def safe_tool(value: int) -> Union[CalculationResult, ErrorResult]: try: if value < 0: raise ValueError("Value must be positive") result = complex_calculation(value) return CalculationResult( success=True, result=result, expression=f"calc({value})", operation="complex_calculation" ) except Exception as e: return ErrorResult( success=False, error_type=type(e).__name__, error_message=str(e), error_code="CALC_ERROR" ) ``` ### 3. 版本兼容性 - **渐进式升级**：逐步迁移现有工具到结构化输出 - **保持向后兼容**：确保现有客户端仍能正常工作 - **文档更新**：及时更新 API 文档和示例 ## 🔧 开发工具 ### 1. 模型验证器 ```python # scripts/validate_schemas.py def validate_tool_schemas(): """验证所有工具的 outputSchema""" for tool in get_all_tools(): if hasattr(tool, 'outputSchema'): validate_json_schema(tool.outputSchema) print(f"✅ {tool.name} schema is valid") ``` ### 2. Schema 生成器 ```python # scripts/generate_docs.py def generate_schema_docs(): """生成 outputSchema 文档""" for tool in get_all_tools(): if hasattr(tool, 'outputSchema'): generate_markdown_doc(tool.name, tool.outputSchema) ``` ## 🚀 后续计划 ### 短期目标 - [ ] 完成所有现有工具的迁移 - [ ] 添加更多响应模型类型 - [ ] 完善测试覆盖率 ### 长期目标 - [ ] 自动化模型生成工具 - [ ] 可视化 Schema 编辑器 - [ ] 客户端 SDK 生成 ## 📞 支持 如果在使用 outputSchema 功能时遇到问题： 1. 查看 [测试文件](../tests/test_output_schema.py) 获取使用示例 2. 参考 [规划文档](OUTPUTSCHEMA_UPGRADE_PLAN.md) 了解实现细节 3. 提交 [GitHub Issue](https://github.com/WW-AI-Lab/Awesome-MCP-Scaffold/issues) 报告问题 --- **💡 提示**：outputSchema 功能完全向后兼容，现有的工具和客户端无需任何修改即可继续正常工作。