工具编排：MCP模型上下文协议

import jsonfrom typing import List, Dict, Any, Optional, Callablefrom dataclasses import dataclass, field  @dataclassclass MCPTool:    """MCP工具定义"""    name: str    description: str    input_schema: Dict[str, Any]    handler: Optional[Callable] = None     def to_schema(self) -> Dict:        return {            "name": self.name,            "description": self.description,            "inputSchema": self.input_schema        }  @dataclassclass MCPResource:    """MCP资源定义"""    uri: str    name: str    description: str    mime_type: str = "text/plain"    content: str = ""  @dataclassclass ToolCallRequest:    """工具调用请求"""    tool_name: str    arguments: Dict[str, Any]  @dataclassclass ToolCallResponse:    """工具调用响应"""    success: bool    result: Any = None    error: str = ""  class MCPServer:    """MCP服务器"""     def __init__(self, name: str, version: str = "1.0.0"):        self.name = name        self.version = version        self.tools: Dict[str, MCPTool] = {}        self.resources: Dict[str, MCPResource] = {}        self.call_log: List[Dict] = []     def register_tool(self, tool: MCPTool):        self.tools[tool.name] = tool     def register_resource(self, resource: MCPResource):        self.resources[resource.uri] = resource     def list_tools(self) -> List[Dict]:        return [tool.to_schema() for tool in self.tools.values()]     def list_resources(self) -> List[Dict]:        return [            {"uri": r.uri, "name": r.name,             "description": r.description, "mimeType": r.mime_type}            for r in self.resources.values()        ]     def read_resource(self, uri: str) -> Optional[str]:        resource = self.resources.get(uri)        if resource is None:            return None        return resource.content     def call_tool(self, request: ToolCallRequest) -> ToolCallResponse:        log_entry = {            "tool": request.tool_name,            "arguments": request.arguments        }        tool = self.tools.get(request.tool_name)        if tool is None:            resp = ToolCallResponse(False, error=f"Tool not found: {request.tool_name}")            log_entry["result"] = "tool_not_found"            self.call_log.append(log_entry)            return resp         valid, error = self._validate_params(request.arguments, tool.input_schema)        if not valid:            resp = ToolCallResponse(False, error=f"Validation: {error}")            log_entry["result"] = f"validation_error: {error}"            self.call_log.append(log_entry)            return resp         try:            if tool.handler:                result = tool.handler(**request.arguments)            else:                result = {"status": "ok", "args": request.arguments}            resp = ToolCallResponse(True, result=result)            log_entry["result"] = "success"        except Exception as e:            resp = ToolCallResponse(False, error=str(e))            log_entry["result"] = f"error: {str(e)}"         self.call_log.append(log_entry)        return resp     def _validate_params(self, args: Dict, schema: Dict) -> tuple:        props = schema.get("properties", {})        required = schema.get("required", [])        for req in required:            if req not in args:                return False, f"Missing: {req}"        for key in args:            if key not in props:                return False, f"Unknown: {key}"        return True, ""     def get_audit_log(self) -> List[Dict]:        return self.call_log  class MCPClient:    """MCP客户端"""     def __init__(self):        self.connected_servers: Dict[str, MCPServer] = {}        self.discovered_tools: Dict[str, Dict] = {}        self.discovered_resources: Dict[str, Dict] = {}     def connect(self, server: MCPServer):        self.connected_servers[server.name] = server        for tool_schema in server.list_tools():            self.discovered_tools[tool_schema["name"]] = {                "server": server.name, **tool_schema            }        for resource_schema in server.list_resources():            self.discovered_resources[resource_schema["uri"]] = {                "server": server.name, **resource_schema            }     def call_tool(self, tool_name: str,                  arguments: Dict) -> ToolCallResponse:        tool_info = self.discovered_tools.get(tool_name)        if tool_info is None:            return ToolCallResponse(False, error="Tool not discovered")        server = self.connected_servers[tool_info["server"]]        request = ToolCallRequest(tool_name, arguments)        return server.call_tool(request)     def read_resource(self, uri: str) -> Optional[str]:        resource_info = self.discovered_resources.get(uri)        if resource_info is None:            return None        server = self.connected_servers[resource_info["server"]]        return server.read_resource(uri)  def run_mcp_demo():    print("=== MCP Protocol Demo ===")     fs_server = MCPServer("filesystem-server", "1.0.0")     fs_server.register_tool(MCPTool(        name="read_file",        description="Read content of a file",        input_schema={            "type": "object",            "properties": {"path": {"type": "string"}},            "required": ["path"]        },        handler=lambda path: f"Content of {path}: Hello World"    ))     fs_server.register_tool(MCPTool(        name="list_directory",        description="List files in a directory",        input_schema={            "type": "object",            "properties": {"path": {"type": "string"}},            "required": ["path"]        },        handler=lambda path: ["file1.txt", "file2.py", "README.md"]    ))     fs_server.register_resource(MCPResource(        uri="file:///project/README.md",        name="Project README",        description="Project documentation file",        content="# My Project\nA sample project for MCP demo."    ))     db_server = MCPServer("database-server", "1.0.0")     db_server.register_tool(MCPTool(        name="query_database",        description="Execute SQL query",        input_schema={            "type": "object",            "properties": {                "sql": {"type": "string"},                "database": {"type": "string"}            },            "required": ["sql"]        },        handler=lambda sql, database="default": {            "rows": [{"id": 1, "name": "Alice"}],            "count": 1        }    ))     db_server.register_resource(MCPResource(        uri="db:///schema/users",        name="Users Table Schema",        description="Database schema for users table",        content="users(id INT, name VARCHAR(100), email VARCHAR(255))"    ))     client = MCPClient()    client.connect(fs_server)    client.connect(db_server)     print(f"\nDiscovered tools: {list(client.discovered_tools.keys())}")    print(f"Discovered resources: {list(client.discovered_resources.keys())}")     print("\n--- Tool Call: list_directory ---")    resp = client.call_tool("list_directory", {"path": "/project"})    print(f"  Success: {resp.success}, Result: {resp.result}")     print("\n--- Tool Call: query_database ---")    resp = client.call_tool("query_database", {"sql": "SELECT * FROM users"})    print(f"  Success: {resp.success}, Result: {resp.result}")     print("\n--- Resource Read: README ---")    content = client.read_resource("file:///project/README.md")    print(f"  Content: {content}")     print("\n--- Resource Read: DB Schema ---")    schema = client.read_resource("db:///schema/users")    print(f"  Schema: {schema}")     print("\n--- Audit Log (filesystem-server) ---")    for entry in fs_server.get_audit_log():        print(f"  {entry}")     print("\n--- Audit Log (database-server) ---")    for entry in db_server.get_audit_log():        print(f"  {entry}")  if __name__ == "__main__":    run_mcp_demo()