MCP 在信息安全平台中的应用：构建智能威胁响应体系

# mcp_iot_device_control_framework.py - MCP IoT 设备控制框架核心实现

from typing import Dict, List, Any, Optional
import json
import time
import threading
from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum

class DeviceStatus(Enum):
    """设备状态枚举"""
    ONLINE = "online"
    OFFLINE = "offline"
    ERROR = "error"
    MAINTENANCE = "maintenance"

class DeviceType(Enum):
    """设备类型枚举"""
    SENSOR = "sensor"
    ACTUATOR = "actuator"
    CONTROLLER = "controller"
    GATEWAY = "gateway"
    ROBOT = "robot"

@dataclass
class PropertyDef:
    """设备属性定义"""
    name: str
    type: str
    unit: Optional[str] = None
    range: Optional[List[Any]] = None
    readable: bool = True
    writable: bool = False
    default_value: Any = None

@dataclass
class CommandDef:
    """设备命令定义"""
    name: str
    description: str
    parameters: Dict[str, Any]
    returns: Dict[str, Any]

class IoT_Device(ABC):
    """IoT 设备抽象基类"""
    
    def __init__(self, device_id: str, device_type: DeviceType, manufacturer: str, model: str):
        self.device_id = device_id
        self.device_type = device_type
        self.manufacturer = manufacturer
        self.model = model
        self.status = DeviceStatus.ONLINE
        self.properties: Dict[str, Any] = {}
        self.property_defs: Dict[str, PropertyDef] = {}
        self.commands: Dict[str, CommandDef] = {}
        self.last_heartbeat = time.time()
        self.lock = threading.Lock()
        self.initialize_properties()
        self.initialize_commands()
    
    @abstractmethod
    def initialize_properties(self) -> None:
        """初始化设备属性定义"""
        pass
    
    @abstractmethod
    def initialize_commands(self) -> None:
        """初始化设备命令定义"""
        pass
    
    @abstractmethod
    def read_property(self, property_name: str) -> Any:
        """读取设备属性"""
        pass
    
    @abstractmethod
    def write_property(self, property_name: str, value: Any) -> bool:
        """写入设备属性"""
        pass
    
    @abstractmethod
    def execute_command(self, command_name: str, parameters: Dict[str, Any]) -> Dict[str, Any]:
        """执行设备命令"""
        pass
    
    def get_device_info(self) -> Dict[str, Any]:
        """获取设备完整信息"""
        with self.lock:
            return {
                "device_id": self.device_id,
                "device_type": self.device_type.value,
                "manufacturer": self.manufacturer,
                "model": self.model,
                "status": self.status.value,
                "properties": self.properties,
                "property_defs": {k: v.__dict__ for k, v in self.property_defs.items()},
                "commands": {k: v.__dict__ for k, v in self.commands.items()},
                "last_heartbeat": self.last_heartbeat,
                "uptime": time.time() - self.last_heartbeat
            }
    
    def heartbeat(self) -> None:
        """设备心跳更新"""
        with self.lock:
            self.last_heartbeat = time.time()
    
    def set_status(self, status: DeviceStatus) -> None:
        """设置设备状态"""
        with self.lock:
            self.status = status

class TemperatureSensor(IoT_Device):
    """温度传感器具体实现"""
    
    def __init__(self, device_id: str, manufacturer: str, model: str):
        super().__init__(device_id, DeviceType.SENSOR, manufacturer, model)
    
    def initialize_properties(self) -> None:
        """初始化温度传感器属性定义"""
        # 温度属性
        self.property_defs["temperature"] = PropertyDef(
            name="temperature",
            type="float",
            unit="°C",
            range=[-50.0, 150.0],
            readable=True,
            writable=False,
            default_value=25.0
        )
        # 湿度属性
        self.property_defs["humidity"] = PropertyDef(
            name="humidity",
            type="float",
            unit="%",
            range=[0.0, 100.0],
            readable=True,
            writable=False,
            default_value=50.0
        )
        # 更新频率属性
        self.property_defs["update_frequency"] = PropertyDef(
            name="update_frequency",
            type="int",
            unit="Hz",
            range=[1, 100],
            readable=True,
            writable=True,
            default_value=10
        )
        
        # 初始化属性值
        for prop_name, prop_def in self.property_defs.items():
            self.properties[prop_name] = prop_def.default_value
    
    def initialize_commands(self) -> None:
        """初始化温度传感器命令定义"""
        # 校准命令
        self.commands["calibrate"] = CommandDef(
            name="calibrate",
            description="校准温度传感器",
            parameters={
                "offset": {
                    "type": "float",
                    "description": "校准偏移量",
                    "required": True,
                    "range": [-10.0, 10.0]
                }
            },
            returns={
                "status": "string",
                "message": "string",
                "calibrated_value": "float"
            }
        )
        # 自检命令
        self.commands["self_test"] = CommandDef(
            name="self_test",
            description="执行传感器自检",
            parameters={},
            returns={
                "status": "string",
                "result": "string",
                "test_details": "object"
            }
        )
    
    def read_property(self, property_name: str) -> Any:
        """读取温度传感器属性"""
        if property_name not in self.property_defs:
            raise ValueError(f"Property {property_name} not found")
        
        with self.lock:
            if property_name == "temperature":
                # 模拟温度传感器读数变化（实际从硬件读取）
                # 加入随机波动，模拟真实环境
                self.properties["temperature"] += (time.time() % 10 - 5) * 0.1
                # 确保温度在范围内
                temp_range = self.property_defs["temperature"].range
                self.properties["temperature"] = max(temp_range[0], min(temp_range[1], self.properties["temperature"]))
            elif property_name == "humidity":
                # 模拟湿度变化
                self.properties["humidity"] += (time.time() % 8 - 4) * 0.5
                # 确保湿度在范围内
                hum_range = self.property_defs["humidity"].range
                self.properties["humidity"] = max(hum_range[0], min(hum_range[1], self.properties["humidity"]))
        
        return self.properties[property_name]
    
    def write_property(self, property_name: str, value: Any) -> bool:
        """写入温度传感器属性"""
        if property_name not in self.property_defs:
            return False
        
        prop_def = self.property_defs[property_name]
        if not prop_def.writable:
            return False
        
        # 验证值是否在范围内
        if prop_def.range:
            if value < prop_def.range[0] or value > prop_def.range[1]:
                return False
        
        with self.lock:
            self.properties[property_name] = value
        return True
    
    def execute_command(self, command_name: str, parameters: Dict[str, Any]) -> Dict[str, Any]:
        """执行温度传感器命令"""
        if command_name not in self.commands:
            raise ValueError(f"Command {command_name} not found")
        
        with self.lock:
            if command_name == "calibrate":
                offset = parameters.get("offset", 0.0)
                # 模拟校准过程
                time.sleep(0.5)  # 模拟校准延迟
                self.properties["temperature"] += offset
                return {
                    "status": "success",
                    "message": f"Sensor calibrated with offset {offset}°C",
                    "calibrated_value": self.properties["temperature"]
                }
            elif command_name == "self_test":
                # 模拟自检过程
                time.sleep(1.0)  # 模拟自检延迟
                return {
                    "status": "success",
                    "result": "passed",
                    "test_details": {
                        "temperature_sensor": "ok",
                        "humidity_sensor": "ok",
                        "communication": "ok",
                        "battery": "95%"
                    }
                }
            return {"status": "error", "message": "Unknown command"}

class DeviceManager:
    """设备管理器，负责设备的注册、发现、管理和控制"""
    
    def __init__(self):
        self.devices: Dict[str, IoT_Device] = {}
        self.lock = threading.Lock()
        self.heartbeat_timeout = 300  # 心跳超时时间（秒）
        self.heartbeat_monitor = threading.Thread(target=self._heartbeat_monitor, daemon=True)
        self.heartbeat_monitor.start()
    
    def _heartbeat_monitor(self):
        """心跳监控线程，定期检查设备状态"""
        while True:
            time.sleep(60)  # 每分钟检查一次
            current_time = time.time()
            with self.lock:
                for device_id, device in self.devices.items():
                    if current_time - device.last_heartbeat > self.heartbeat_timeout:
                        device.set_status(DeviceStatus.OFFLINE)
    
    def register_device(self, device: IoT_Device) -> bool:
        """注册设备"""
        with self.lock:
            if device.device_id not in self.devices:
                self.devices[device.device_id] = device
                return True
            return False
    
    def unregister_device(self, device_id: str) -> bool:
        """注销设备"""
        with self.lock:
            if device_id in self.devices:
                del self.devices[device_id]
                return True
            return False
    
    def list_devices(self, device_type: Optional[DeviceType] = None) -> List[Dict[str, Any]]:
        """列出所有设备"""
        with self.lock:
            result = []
            for device in self.devices.values():
                if device_type is None or device.device_type == device_type:
                    result.append(device.get_device_info())
            return result
    
    def get_device(self, device_id: str) -> Optional[IoT_Device]:
        """获取设备实例"""
        with self.lock:
            return self.devices.get(device_id)
    
    def read_device_property(self, device_id: str, property_name: str) -> Any:
        """读取设备属性"""
        device = self.get_device(device_id)
        if not device:
            raise ValueError(f"Device {device_id} not found")
        return device.read_property(property_name)
    
    def write_device_property(self, device_id: str, property_name: str, value: Any) -> bool:
        """写入设备属性"""
        device = self.get_device(device_id)
        if not device:
            return False
        return device.write_property(property_name, value)
    
    def execute_device_command(self, device_id: str, command_name: str, parameters: Dict[str, Any]) -> Dict[str, Any]:
        """执行设备命令"""
        device = self.get_device(device_id)
        if not device:
            raise ValueError(f"Device {device_id} not found")
        return device.execute_command(command_name, parameters)
    
    def update_device_heartbeat(self, device_id: str) -> bool:
        """更新设备心跳"""
        device = self.get_device(device_id)
        if not device:
            return False
        device.heartbeat()
        device.set_status(DeviceStatus.ONLINE)
        return True

# MCP IoT 设备控制框架主类
class MCPIoTDeviceControlFramework:
    """MCP IoT 设备控制框架主类"""
    
    def __init__(self):
        self.device_manager = DeviceManager()
        self.security_manager = self._init_security_manager()
        self.monitoring_manager = self._init_monitoring_manager()
        self.audit_manager = self._init_audit_manager()
    
    def _init_security_manager(self):
        """初始化安全管理器"""
        # 实际实现会包含认证、授权、加密等功能
        return type('SecurityManager', (), {})
    
    def _init_monitoring_manager(self):
        """初始化监控管理器"""
        # 实际实现会包含实时监控、告警等功能
        return type('MonitoringManager', (), {})
    
    def _init_audit_manager(self):
        """初始化审计管理器"""
        # 实际实现会包含日志记录、审计等功能
        return type('AuditManager', (), {})
    
    def register_device(self, device: IoT_Device) -> Dict[str, Any]:
        """注册设备到 MCP 框架"""
        success = self.device_manager.register_device(device)
        if success:
            self.audit_manager.log_audit(
                action="register_device",
                device_id=device.device_id,
                details={"manufacturer": device.manufacturer, "model": device.model}
            )
        return {"success": success, "device_id": device.device_id}
    
    def get_devices(self, device_type: Optional[str] = None) -> List[Dict[str, Any]]:
        """获取所有设备列表"""
        device_type_enum = None
        if device_type:
            device_type_enum = DeviceType(device_type)
        return self.device_manager.list_devices(device_type_enum)
    
    def call_device(self, device_id: str, action: str, params: Dict[str, Any]) -> Dict[str, Any]:
        """调用设备操作"""
        if action == "read_property":
            property_name = params.get("property_name")
            if not property_name:
                return {"success": False, "error": "property_name is required"}
            try:
                value = self.device_manager.read_device_property(device_id, property_name)
                self.audit_manager.log_audit(
                    action="read_property",
                    device_id=device_id,
                    details={"property": property_name, "value": value}
                )
                return {"success": True, "value": value}
            except Exception as e:
                return {"success": False, "error": str(e)}
        elif action == "write_property":
            property_name = params.get("property_name")
            value = params.get("value")
            if not property_name or value is None:
                return {"success": False, "error": "property_name and value are required"}
            success = self.device_manager.write_device_property(device_id, property_name, value)
            if success:
                self.audit_manager.log_audit(
                    action="write_property",
                    device_id=device_id,
                    details={"property": property_name, "value": value}
                )
            return {"success": success}
        elif action == "execute_command":
            command_name = params.get("command_name")
            command_params = params.get("params", {})
            if not command_name:
                return {"success": False, "error": "command_name is required"}
            try:
                result = self.device_manager.execute_device_command(device_id, command_name, command_params)
                self.audit_manager.log_audit(
                    action="execute_command",
                    device_id=device_id,
                    details={"command": command_name, "params": command_params, "result": result}
                )
                return {"success": True, "result": result}
            except Exception as e:
                return {"success": False, "error": str(e)}
        else:
            return {"success": False, "error": f"Unknown action: {action}"}

# 使用示例
if __name__ == "__main__":
    # 初始化 MCP IoT 设备控制框架
    framework = MCPIoTDeviceControlFramework()
    
    # 注册温度传感器
    temp_sensor = TemperatureSensor("temp-sensor-001", "Industrial Sensors Inc.", "TS-1000")
    framework.register_device(temp_sensor)
    
    # 注册第二个温度传感器
    temp_sensor2 = TemperatureSensor("temp-sensor-002", "Industrial Sensors Inc.", "TS-2000")
    framework.register_device(temp_sensor2)
    
    # 列出所有设备
    print("=== 所有设备列表 ===")
    devices = framework.get_devices()
    for device in devices:
        print(f"设备 ID: {device['device_id']}, 类型: {device['device_type']}, 状态: {device['status']}")
        print(f"  制造商: {device['manufacturer']}, 型号: {device['model']}")
        print(f"  当前温度: {device['properties'].get('temperature'):.2f}°C")
        print(f"  当前湿度: {device['properties'].get('humidity'):.2f}%")
    
    # 读取设备属性
    print("\n=== 读取设备属性 ===")
    result = framework.call_device("temp-sensor-001", "read_property", {"property_name": "temperature"})
    print(f"读取温度传感器 001 温度: {result['value']:.2f}°C")
    
    # 修改设备属性
    print("\n=== 修改设备属性 ===")
    result = framework.call_device(
        "temp-sensor-001", 
        "write_property", 
        {"property_name": "update_frequency", "value": 20}
    )
    print(f"修改温度传感器 001 更新频率: {'成功' if result['success'] else '失败'}")
    
    # 执行设备命令
    print("\n=== 执行设备命令 ===")
    result = framework.call_device(
        "temp-sensor-001", 
        "execute_command", 
        {"command_name": "calibrate", "params": {"offset": -1.5}}
    )
    print(f"校准温度传感器 001: {result['result']['message']}")
    
    # 执行自检命令
    print("\n=== 执行设备自检 ===")
    result = framework.call_device(
        "temp-sensor-001", 
        "execute_command", 
        {"command_name": "self_test", "params": {}}
    )
    print(f"传感器自检结果: {result['result']['status']} - {result['result']['result']}")

# 安装依赖
pip install requests

# 运行示例代码
python mcp_iot_device_control_framework.py

=== 所有设备列表 ===
设备 ID: temp-sensor-001, 类型: sensor, 状态: online
  制造商: Industrial Sensors Inc., 型号: TS-1000
  当前温度: 25.00°C
  当前湿度: 50.00%
设备 ID: temp-sensor-002, 类型: sensor, 状态: online
  制造商: Industrial Sensors Inc., 型号: TS-2000
  当前温度: 25.00°C
  当前湿度: 50.00%

=== 读取设备属性 ===
读取温度传感器 001 温度: 24.50°C

=== 修改设备属性 ===
修改温度传感器 001 更新频率: 成功

=== 执行设备命令 ===
校准温度传感器 001: Sensor calibrated with offset -1.5°C

=== 执行设备自检 ===
传感器自检结果: success - passed

# mcp_industrial_protocol_adapter.py - MCP 工业实时通信协议适配器

from typing import Dict, List, Any, Optional
import json
import time
from abc import ABC, abstractmethod
from enum import Enum
from threading import Thread, Lock

class IndustrialProtocol(Enum):
    """工业通信协议枚举"""
    MODBUS = "modbus"
    OPCUA = "opcua"
    PROFINET = "profinet"
    ETHERNET_IP = "ethernet_ip"
    MQTT = "mqtt"

class ProtocolAdapter(ABC):
    """工业协议适配器抽象基类"""
    
    def __init__(self, protocol: IndustrialProtocol, config: Dict[str, Any]):
        self.protocol = protocol
        self.config = config
        self.running = False
        self.lock = Lock()
        self.event_handlers = []
    
    @abstractmethod
    def connect(self) -> bool:
        """连接到协议栈"""
        pass
    
    @abstractmethod
    def disconnect(self) -> bool:
        """断开连接"""
        pass
    
    @abstractmethod
    def read(self, address: str, params: Dict[str, Any]) -> Dict[str, Any]:
        """读取数据"""
        pass
    
    @abstractmethod
    def write(self, address: str, params: Dict[str, Any]) -> Dict[str, Any]:
        """写入数据"""
        pass
    
    @abstractmethod
    def subscribe(self, address: str, params: Dict[str, Any]) -> bool:
        """订阅数据"""
        pass
    
    @abstractmethod
    def unsubscribe(self, address: str) -> bool:
        """取消订阅"""
        pass
    
    def register_event_handler(self, handler):
        """注册事件处理器"""
        with self.lock:
            if handler not in self.event_handlers:
                self.event_handlers.append(handler)
    
    def unregister_event_handler(self, handler):
        """注销事件处理器"""
        with self.lock:
            if handler in self.event_handlers:
                self.event_handlers.remove(handler)
    
    def notify_event(self, event: Dict[str, Any]):
        """通知事件"""
        with self.lock:
            for handler in self.event_handlers:
                try:
                    handler(event)
                except Exception as e:
                    print(f"Error handling event: {e}")

class ModbusAdapter(ProtocolAdapter):
    """Modbus 协议适配器实现"""
    
    def __init__(self, config: Dict[str, Any]):
        super().__init__(IndustrialProtocol.MODBUS, config)
        # 模拟 Modbus 客户端，实际实现会使用 pymodbus 等库
        self.modbus_client = None
    
    def connect(self) -> bool:
        """连接到 Modbus 设备"""
        # 模拟连接过程
        print(f"Connecting to Modbus device at {self.config.get('host')}:{self.config.get('port')}")
        time.sleep(0.5)
        self.running = True
        return True
    
    def disconnect(self) -> bool:
        """断开 Modbus 连接"""
        print(f"Disconnecting from Modbus device")
        time.sleep(0.2)
        self.running = False
        return True
    
    def read(self, address: str, params: Dict[str, Any]) -> Dict[str, Any]:
        """读取 Modbus 寄存器"""
        # 模拟 Modbus 读取
        register_type = params.get("register_type", "holding")
        count = params.get("count", 1)
        
        # 模拟读取结果
        values = [i + int(time.time() % 100) for i in range(count)]
        
        return {
            "success": True,
            "address": address,
            "values": values,
            "timestamp": time.time()
        }
    
    def write(self, address: str, params: Dict[str, Any]) -> Dict[str, Any]:
        """写入 Modbus 寄存器"""
        # 模拟 Modbus 写入
        values = params.get("values", [])
        register_type = params.get("register_type", "holding")
        
        print(f"Writing to Modbus register {address} values: {values}")
        time.sleep(0.3)  # 模拟写入延迟
        
        return {
            "success": True,
            "address": address,
            "written_values": values,
            "timestamp": time.time()
        }
    
    def subscribe(self, address: str, params: Dict[str, Any]) -> bool:
        """订阅 Modbus 数据变化"""
        print(f"Subscribing to Modbus address {address}")
        return True
    
    def unsubscribe(self, address: str) -> bool:
        """取消订阅"""
        print(f"Unsubscribing from Modbus address {address}")
        return True

class OPCUAAdapter(ProtocolAdapter):
    """OPC UA 协议适配器实现"""
    
    def __init__(self, config: Dict[str, Any]):
        super().__init__(IndustrialProtocol.OPCUA, config)
        # 模拟 OPC UA 客户端，实际实现会使用 opcua 等库
        self.opcua_client = None
    
    def connect(self) -> bool:
        """连接到 OPC UA 服务器"""
        print(f"Connecting to OPC UA server at {self.config.get('url')}")
        time.sleep(1.0)  # 模拟 OPC UA 连接延迟
        self.running = True
        return True
    
    def disconnect(self) -> bool:
        """断开 OPC UA 连接"""
        print(f"Disconnecting from OPC UA server")
        time.sleep(0.5)
        self.running = False
        return True
    
    def read(self, address: str, params: Dict[str, Any]) -> Dict[str, Any]:
        """读取 OPC UA 节点"""
        print(f"Reading OPC UA node {address}")
        time.sleep(0.8)  # 模拟 OPC UA 读取延迟
        
        return {
            "success": True,
            "node_id": address,
            "value": 123.45,
            "timestamp": time.time()
        }
    
    def write(self, address: str, params: Dict[str, Any]) -> Dict[str, Any]:
        """写入 OPC UA 节点"""
        value = params.get("value")
        print(f"Writing to OPC UA node {address} value: {value}")
        time.sleep(1.2)  # 模拟 OPC UA 写入延迟
        
        return {
            "success": True,
            "node_id": address,
            "written_value": value,
            "timestamp": time.time()
        }
    
    def subscribe(self, address: str, params: Dict[str, Any]) -> bool:
        """订阅 OPC UA 节点变化"""
        print(f"Subscribing to OPC UA node {address}")
        return True
    
    def unsubscribe(self, address: str) -> bool:
        """取消订阅"""
        print(f"Unsubscribing from OPC UA node {address}")
        return True

class IndustrialProtocolAdapter:
    """MCP 工业协议适配器主类"""
    
    def __init__(self):
        self.adapters = {}
        self.lock = Lock()
    
    def register_adapter(self, protocol: IndustrialProtocol, adapter: ProtocolAdapter) -> bool:
        """注册协议适配器"""
        with self.lock:
            if protocol not in self.adapters:
                self.adapters[protocol] = adapter
                return adapter.connect()
            return False
    
    def get_adapter(self, protocol: IndustrialProtocol) -> Optional[ProtocolAdapter]:
        """获取协议适配器"""
        with self.lock:
            return self.adapters.get(protocol)
    
    def call_protocol(self, protocol: str, action: str, params: Dict[str, Any]) -> Dict[str, Any]:
        """调用工业协议"""
        try:
            protocol_enum = IndustrialProtocol(protocol)
        except ValueError:
            return {"success": False, "error": f"Unknown protocol: {protocol}"}
        
        adapter = self.get_adapter(protocol_enum)
        if not adapter:
            return {"success": False, "error": f"Adapter not found for protocol: {protocol}"}
        
        address = params.get("address")
        if not address:
            return {"success": False, "error": "address is required"}
        
        try:
            if action == "read":
                return adapter.read(address, params)
            elif action == "write":
                return adapter.write(address, params)
            elif action == "subscribe":
                result = adapter.subscribe(address, params)
                return {"success": result, "address": address}
            elif action == "unsubscribe":
                result = adapter.unsubscribe(address)
                return {"success": result, "address": address}
            else:
                return {"success": False, "error": f"Unknown action: {action}"}
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    def mcp_to_protocol(self, mcp_request: Dict[str, Any]) -> Dict[str, Any]:
        """MCP 请求转换为工业协议请求"""
        # MCP 请求格式：{"tool": "industrial_protocol", "params": {"protocol": "modbus", "action": "read", "address": "192.168.1.100:502/40001", "register_type": "holding", "count": 2}}
        params = mcp_request.get("params", {})
        protocol = params.get("protocol")
        action = params.get("action")
        
        if not protocol or not action:
            return {"success": False, "error": "protocol and action are required"}
        
        result = self.call_protocol(protocol, action, params)
        return result
    
    def protocol_to_mcp(self, protocol_response: Dict[str, Any]) -> Dict[str, Any]:
        """工业协议响应转换为 MCP 响应"""
        # 工业协议响应转换为 MCP 格式
        return {
            "success": protocol_response.get("success", False),
            "result": protocol_response,
            "timestamp": time.time()
        }

# 使用示例
if __name__ == "__main__":
    # 初始化工业协议适配器
    adapter_manager = IndustrialProtocolAdapter()
    
    # 注册 Modbus 适配器
    modbus_config = {"host": "192.168.1.100", "port": 502, "slave_id": 1}
    modbus_adapter = ModbusAdapter(modbus_config)
    adapter_manager.register_adapter(IndustrialProtocol.MODBUS, modbus_adapter)
    
    # 注册 OPC UA 适配器
    opcua_config = {"url": "opc.tcp://192.168.1.200:4840", "namespace": "http://example.com"}
    opcua_adapter = OPCUAAdapter(opcua_config)
    adapter_manager.register_adapter(IndustrialProtocol.OPCUA, opcua_adapter)
    
    # 测试 Modbus 读取
    print("=== 测试 Modbus 读取 ===")
    modbus_request = {
        "tool": "industrial_protocol",
        "params": {
            "protocol": "modbus",
            "action": "read",
            "address": "192.168.1.100:502/40001",
            "register_type": "holding",
            "count": 2
        }
    }
    result = adapter_manager.mcp_to_protocol(modbus_request)
    print(f"Modbus 读取结果: {result}")
    
    # 测试 Modbus 写入
    print("\n=== 测试 Modbus 写入 ===")
    modbus_write_request = {
        "tool": "industrial_protocol",
        "params": {
            "protocol": "modbus",
            "action": "write",
            "address": "192.168.1.100:502/40001",
            "register_type": "holding",
            "values": [123, 456]
        }
    }
    result = adapter_manager.mcp_to_protocol(modbus_write_request)
    print(f"Modbus 写入结果: {result}")
    
    # 测试 OPC UA 读取
    print("\n=== 测试 OPC UA 读取 ===")
    opcua_request = {
        "tool": "industrial_protocol",
        "params": {
            "protocol": "opcua",
            "action": "read",
            "address": "ns=2;s=TemperatureSensor",
            "node_id": "ns=2;s=TemperatureSensor"
        }
    }
    result = adapter_manager.mcp_to_protocol(opcua_request)
    print(f"OPC UA 读取结果: {result}")
    
    # 测试 OPC UA 订阅
    print("\n=== 测试 OPC UA 订阅 ===")
    opcua_subscribe_request = {
        "tool": "industrial_protocol",
        "params": {
            "protocol": "opcua",
            "action": "subscribe",
            "address": "ns=2;s=PressureSensor",
            "sampling_interval": 1000
        }
    }
    result = adapter_manager.mcp_to_protocol(opcua_subscribe_request)
    print(f"OPC UA 订阅结果: {result}")

### 3.3 MCP 智能工厂集成方案

MCP 智能工厂集成方案是一个用于将 MCP 集成到智能工厂中的完整解决方案，它实现了与企业现有工业系统的无缝集成。

#### 3.3.1 智能工厂集成架构

```mermaid
graph TD
    subgraph 企业级系统
        A[ERP 系统]
        B[MES 系统]
        C[SCADA 系统]
        D[PLM 系统]
    end
    
    subgraph MCP 集成层
        E[MCP 智能工厂集成平台]
        E --> F[ERP 适配器]
        E --> G[MES 适配器]
        E --> H[SCADA 适配器]
        E --> I[PLM 适配器]
    end
    
    subgraph 生产执行层
        J[PLC 控制器]
        K[机器人系统]
        L[传感器网络]
        M[AGV 系统]
    end
    
    subgraph MCP 设备层
        N[MCP Server]
        N --> O[MCP IoT 设备控制框架]
        N --> P[MCP 工业协议适配器]
    end
    
    A --> F
    B --> G
    C --> H
    D --> I
    
    F --> E
    G --> E
    H --> E
    I --> E
    
    E --> P
    P --> J
    P --> K
    P --> L
    P --> M
    
    O --> L
    
    classDef enterprise fill:#4169E1,stroke:#333,stroke-width:2px;
    classDef mcp_integration fill:#32CD32,stroke:#333,stroke-width:2px;
    classDef production fill:#FFA500,stroke:#333,stroke-width:2px;
    classDef mcp_core fill:#DA70D6,stroke:#333,stroke-width:2px;
    
    class A,B,C,D enterprise;
    class E,F,G,H,I mcp_integration;
    class J,K,L,M production;
    class N,O,P mcp_core;

# mcp_smart_factory_integration.py - MCP 智能工厂集成平台

from typing import Dict, List, Any, Optional
import json
import time
from enum import Enum
from abc import ABC, abstractmethod

class SystemType(Enum):
    """企业系统类型枚举"""
    ERP = "erp"
    MES = "mes"
    SCADA = "scada"
    PLM = "plm"

class EnterpriseSystemAdapter(ABC):
    """企业系统适配器抽象基类"""
    
    def __init__(self, system_type: SystemType, config: Dict[str, Any]):
        self.system_type = system_type
        self.config = config
        self.connected = False
    
    @abstractmethod
    def connect(self) -> bool:
        """连接到企业系统"""
        pass
    
    @abstractmethod
    def disconnect(self) -> bool:
        """断开连接"""
        pass
    
    @abstractmethod
    def call_system(self, action: str, params: Dict[str, Any]) -> Dict[str, Any]:
        """调用企业系统 API"""
        pass

class ERPAdapter(EnterpriseSystemAdapter):
    """ERP 系统适配器实现"""
    
    def __init__(self, config: Dict[str, Any]):
        super().__init__(SystemType.ERP, config)
    
    def connect(self) -> bool:
        """连接到 ERP 系统"""
        print(f"Connecting to ERP system at {self.config.get('url')}")
        time.sleep(1.0)  # 模拟 ERP 连接延迟
        self.connected = True
        return True
    
    def disconnect(self) -> bool:
        """断开 ERP 连接"""
        print("Disconnecting from ERP system")
        time.sleep(0.5)
        self.connected = False
        return True
    
    def call_system(self, action: str, params: Dict[str, Any]) -> Dict[str, Any]:
        """调用 ERP 系统 API"""
        if action == "get_order":
            order_id = params.get("order_id")
            # 模拟获取订单信息
            return {
                "success": True,
                "order_id": order_id,
                "product_id": "PROD-001",
                "quantity": 100,
                "status": "in_production",
                "due_date": "2026-02-01T18:00:00Z",
                "customer": "ABC Corporation"
            }
        elif action == "update_order_status":
            order_id = params.get("order_id")
            status = params.get("status")
            return {
                "success": True,
                "order_id": order_id,
                "new_status": status,
                "timestamp": time.time()
            }
        else:
            return {"success": False, "error": f"Unknown action: {action}"}

class MESAdapter(EnterpriseSystemAdapter):
    """MES 系统适配器实现"""
    
    def __init__(self, config: Dict[str, Any]):
        super().__init__(SystemType.MES, config)
    
    def connect(self) -> bool:
        """连接到 MES 系统"""
        print(f"Connecting to MES system at {self.config.get('url')}")
        time.sleep(0.8)  # 模拟 MES 连接延迟
        self.connected = True
        return True
    
    def disconnect(self) -> bool:
        """断开 MES 连接"""
        print("Disconnecting from MES system")
        time.sleep(0.3)
        self.connected = False
        return True
    
    def call_system(self, action: str, params: Dict[str, Any]) -> Dict[str, Any]:
        """调用 MES 系统 API"""
        if action == "get_production_data":
            production_line = params.get("production_line", "line_1")
            start_time = params.get("start_time")
            end_time = params.get("end_time")
            # 模拟获取生产数据
            return {
                "success": True,
                "production_line": production_line,
                "start_time": start_time,
                "end_time": end_time,
                "total_products": 85,
                "good_products": 82,
                "defect_products": 3,
                "downtime": 15,
                "efficiency": 96.5
            }
        elif action == "update_production_status":
            production_line = params.get("production_line")
            status = params.get("status")
            return {
                "success": True,
                "production_line": production_line,
                "new_status": status,
                "timestamp": time.time()
            }
        else:
            return {"success": False, "error": f"Unknown action: {action}"}

class SCADAAdapter(EnterpriseSystemAdapter):
    """SCADA 系统适配器实现"""
    
    def __init__(self, config: Dict[str, Any]):
        super().__init__(SystemType.SCADA, config)
    
    def connect(self) -> bool:
        """连接到 SCADA 系统"""
        print(f"Connecting to SCADA system at {self.config.get('url')}")
        time.sleep(0.5)  # 模拟 SCADA 连接延迟
        self.connected = True
        return True
    
    def disconnect(self) -> bool:
        """断开 SCADA 连接"""
        print("Disconnecting from SCADA system")
        time.sleep(0.2)
        self.connected = False
        return True
    
    def call_system(self, action: str, params: Dict[str, Any]) -> Dict[str, Any]:
        """调用 SCADA 系统 API"""
        if action == "get_tag_value":
            tag_name = params.get("tag_name")
            # 模拟获取 SCADA 标签值
            return {
                "success": True,
                "tag_name": tag_name,
                "value": 123.45,
                "timestamp": time.time()
            }
        elif action == "set_tag_value":
            tag_name = params.get("tag_name")
            value = params.get("value")
            return {
                "success": True,
                "tag_name": tag_name,
                "set_value": value,
                "timestamp": time.time()
            }
        else:
            return {"success": False, "error": f"Unknown action: {action}"}

class SmartFactoryIntegrationPlatform:
    """MCP 智能工厂集成平台"""
    
    def __init__(self):
        self.adapters = {}
    
    def register_adapter(self, adapter: EnterpriseSystemAdapter) -> bool:
        """注册企业系统适配器"""
        system_type = adapter.system_type
        if system_type not in self.adapters:
            self.adapters[system_type] = adapter
            return adapter.connect()
        return False
    
    def get_adapter(self, system_type: SystemType) -> Optional[EnterpriseSystemAdapter]:
        """获取企业系统适配器"""
        return self.adapters.get(system_type)
    
    def call_enterprise_system(self, system_type: str, action: str, params: Dict[str, Any]) -> Dict[str, Any]:
        """调用企业系统"""
        try:
            system_type_enum = SystemType(system_type)
        except ValueError:
            return {"success": False, "error": f"Unknown system type: {system_type}"}
        
        adapter = self.get_adapter(system_type_enum)
        if not adapter or not adapter.connected:
            return {"success": False, "error": f"Adapter not available for {system_type}"}
        
        try:
            result = adapter.call_system(action, params)
            return result
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    def integrate_production_data(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """集成生产数据"""
        # 从不同系统获取数据并集成
        production_line = params.get("production_line", "line_1")
        
        # 从 MES 获取生产数据
        mes_result = self.call_enterprise_system(
            "mes", 
            "get_production_data", 
            {"production_line": production_line}
        )
        
        # 从 SCADA 获取实时数据
        scada_result = self.call_enterprise_system(
            "scada", 
            "get_tag_value", 
            {"tag_name": f"{production_line}/temperature"}
        )
        
        # 从 ERP 获取订单信息
        erp_result = self.call_enterprise_system(
            "erp", 
            "get_order", 
            {"order_id": params.get("order_id")}
        )
        
        # 集成数据
        integrated_data = {
            "production_line": production_line,
            "order_info": erp_result.get("result", {}),
            "production_stats": mes_result.get("result", {}),
            "real_time_data": {
                "temperature": scada_result.get("result", {}).get("value", 0),
                "timestamp": time.time()
            }
        }
        
        return {"success": True, "integrated_data": integrated_data}

# 使用示例
if __name__ == "__main__":
    # 初始化智能工厂集成平台
    platform = SmartFactoryIntegrationPlatform()
    
    # 注册 ERP 适配器
    erp_config = {"url": "https://erp.example.com/api", "api_key": "erp-api-key"}
    erp_adapter = ERPAdapter(erp_config)
    platform.register_adapter(erp_adapter)
    
    # 注册 MES 适配器
    mes_config = {"url": "https://mes.example.com/api", "api_key": "mes-api-key"}
    mes_adapter = MESAdapter(mes_config)
    platform.register_adapter(mes_adapter)
    
    # 注册 SCADA 适配器
    scada_config = {"url": "https://scada.example.com/api", "api_key": "scada-api-key"}
    scada_adapter = SCADAAdapter(scada_config)
    platform.register_adapter(scada_adapter)
    
    # 调用 ERP 系统
    print("=== 调用 ERP 系统 ===")
    result = platform.call_enterprise_system(
        "erp", 
        "get_order", 
        {"order_id": "ORD-2026-0001"}
    )
    print(f"获取订单信息: {result}")
    
    # 调用 MES 系统
    print("\n=== 调用 MES 系统 ===")
    result = platform.call_enterprise_system(
        "mes", 
        "get_production_data", 
        {"production_line": "line_1"}
    )
    print(f"获取生产线数据: {result}")
    
    # 调用 SCADA 系统
    print("\n=== 调用 SCADA 系统 ===")
    result = platform.call_enterprise_system(
        "scada", 
        "get_tag_value", 
        {"tag_name": "line_1/temperature"}
    )
    print(f"获取温度数据: {result}")
    
    # 集成生产数据
    print("\n=== 集成生产数据 ===")
    result = platform.integrate_production_data(
        {"production_line": "line_1", "order_id": "ORD-2026-0001"}
    )
    print(f"集成数据: {json.dumps(result['integrated_data'], indent=2)}")
    
    # 更新生产状态
    print("\n=== 更新生产状态 ===")
    result = platform.call_enterprise_system(
        "mes", 
        "update_production_status", 
        {"production_line": "line_1", "status": "paused"}
    )
    print(f"更新生产线状态: {'成功' if result['success'] else '失败'}")