feat(tool system): add mcp testing services

2025-12-30 10:00:37 +08:00
parent 7f823ee72e
commit 8e893662f3
5 changed files with 364 additions and 0 deletions
--- a/mcp_base.py
+++ b/mcp_base.py
@@ -0,0 +1,111 @@
+#!/usr/bin/env python3
+"""MCP服务器基础模块 - 共享的模型和处理逻辑"""
+
+from pydantic import BaseModel
+from typing import Dict, Any
+
+class MCPRequest(BaseModel):
+    jsonrpc: str = "2.0"
+    id: str
+    method: str
+    params: Dict[str, Any] = {}
+
+class MCPResponse(BaseModel):
+    jsonrpc: str = "2.0"
+    id: str
+    result: Any = None
+    error: Dict[str, Any] = None
+
+# 工具定义
+TOOLS = [
+    {
+        "name": "calculator",
+        "description": "简单计算器",
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "expression": {"type": "string", "description": "数学表达式"}
+            },
+            "required": ["expression"]
+        }
+    },
+    {
+        "name": "echo",
+        "description": "回显工具",
+        "inputSchema": {
+            "type": "object", 
+            "properties": {
+                "message": {"type": "string", "description": "要回显的消息"}
+            },
+            "required": ["message"]
+        }
+    }
+]
+
+async def handle_mcp_request(request: MCPRequest, server_name: str = "MCP Server"):
+    """处理MCP请求"""
+    try:
+        if request.method == "initialize":
+            return MCPResponse(
+                id=request.id,
+                result={
+                    "protocolVersion": "2024-11-05",
+                    "capabilities": {"tools": {"listChanged": True}},
+                    "serverInfo": {"name": server_name, "version": "1.0.0"}
+                }
+            )
+        
+        elif request.method == "tools/list":
+            return MCPResponse(
+                id=request.id,
+                result={"tools": TOOLS}
+            )
+        
+        elif request.method == "tools/call":
+            tool_name = request.params.get("name")
+            arguments = request.params.get("arguments", {})
+            
+            if tool_name == "calculator":
+                try:
+                    expression = arguments.get("expression", "")
+                    result = eval(expression)
+                    return MCPResponse(
+                        id=request.id,
+                        result={"content": [{"type": "text", "text": f"结果: {result}"}]}
+                    )
+                except Exception as e:
+                    return MCPResponse(
+                        id=request.id,
+                        error={"code": -1, "message": f"计算错误: {str(e)}"}
+                    )
+            
+            elif tool_name == "echo":
+                message = arguments.get("message", "")
+                return MCPResponse(
+                    id=request.id,
+                    result={"content": [{"type": "text", "text": f"Echo: {message}"}]}
+                )
+            
+            else:
+                return MCPResponse(
+                    id=request.id,
+                    error={"code": -1, "message": f"未知工具: {tool_name}"}
+                )
+        
+        elif request.method == "ping":
+            return MCPResponse(
+                id=request.id,
+                result={"status": "pong"}
+            )
+        
+        else:
+            return MCPResponse(
+                id=request.id,
+                error={"code": -1, "message": f"未知方法: {request.method}"}
+            )
+    
+    except Exception as e:
+        return MCPResponse(
+            id=request.id,
+            error={"code": -1, "message": str(e)}
+        )