供应链智能体MCP多ERP集成 — Model Context Protocol驱动的多系统双向协调模式

import json
import time
from typing import Any, Dict, List, Optional, Callable
from dataclasses import dataclass, field

@dataclass
class MCPTool:
    """MCP 工具定义"""
    name: str
    description: str
    input_schema: Dict
    handler: Callable  # 实际执行函数

@dataclass
class MCPCallResult:
    """MCP 调用结果"""
    tool_name: str
    success: bool
    data: Any = None
    error: Optional[str] = None
    latency_ms: float = 0.0
    timestamp: str = field(default_factory=lambda: time.strftime("%Y-%m-%dT%H:%M:%S"))

class MCPServer:
    """
    MCP Server 基类
    每个外部系统（SAP/Oracle/WMS）继承此类实现自己的工具集
    """
    
    def __init__(self, server_name: str, system_type: str):
        self.server_name = server_name
        self.system_type = system_type
        self.tools: Dict[str, MCPTool] = {}
        self.call_log: List[MCPCallResult] = []
    
    def register_tool(self, tool: MCPTool):
        self.tools[tool.name] = tool
    
    def call_tool(self, tool_name: str, arguments: Dict) -> MCPCallResult:
        """统一工具调用入口（带延迟测量和日志）"""
        if tool_name not in self.tools:
            result = MCPCallResult(tool_name, False, 
                                   error=f"工具 '{tool_name}' 不存在于 {self.server_name}")
            self.call_log.append(result)
            return result
        
        t0 = time.time()
        try:
            data = self.tools[tool_name].handler(arguments)
            result = MCPCallResult(tool_name, True, data=data,
                                   latency_ms=round((time.time() - t0) * 1000, 2))
        except Exception as e:
            result = MCPCallResult(tool_name, False, error=str(e),
                                   latency_ms=round((time.time() - t0) * 1000, 2))
        
        self.call_log.append(result)
        return result
    
    def list_tools(self) -> List[Dict]:
        return [{"name": t.name, "description": t.description}