OoderAgent P2P 网络核心技术揭秘：多 Agent 协作入网架构与全链路安全防护

{
  "type": "ROUTE_AGENT_DISCOVERY",
  "version": "1.0",
  "timestamp": "2026-01-27T12:00:00Z",
  "sender": "route_agent_id",
  "payload": {
    "scene_ids": ["scene1", "scene2"],
    "capabilities": ["routing", "security"],
    "mcp_agent_id": "mcp_agent_id"
  },
  "signature": "digital_signature"
}

{
  "type": "SECURITY_KEY_BROADCAST",
  "version": "1.0",
  "timestamp": "2026-01-27T12:00:00Z",
  "sender": "route_agent_id",
  "payload": {
    "group_id": "group_uuid",
    "security_key": "encrypted_security_key",
    "key_expiry": "2026-01-28T12:00:00Z"
  },
  "signature": "digital_signature"
}

{
  "type": "CAP_CREATE",
  "version": "1.0",
  "timestamp": "2026-01-27T12:00:00Z",
  "sender": "end_agent_id",
  "payload": {
    "group_id": "group_uuid",
    "link_info": {
      "route_agent_id": "route_agent_id",
      "ip_address": "192.168.1.100",
      "port": 7777,
      "protocol": "tcp"
    },
    "security_key": "encrypted_security_key"
  },
  "signature": "digital_signature"
}

{
  "type": "MCP_AGENT_JOIN",
  "version": "1.0",
  "timestamp": "2026-01-27T12:00:00Z",
  "sender": "route_agent_id",
  "receiver": "mcp_agent_id",
  "payload": {
    "route_agent_id": "route_agent_id",
    "capabilities": ["routing", "security"],
    "auth_info": "encrypted_auth_info"
  },
  "signature": "digital_signature"
}

package net.ooder.sdk.agent.impl;

import net.ooder.sdk.agent.EndAgent;
import net.ooder.sdk.packet.TaskPacket;
import net.ooder.sdk.packet.ResponsePacket;
import net.ooder.sdk.packet.ResponsePacketBuilder;
import net.ooder.sdk.network.udp.UDPSDK;
import net.ooder.sdk.enums.EndAgentStatus;
import net.ooder.sdk.enums.ResponseStatus;
import net.ooder.sdk.security.EncryptionUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class EndAgentImpl extends AbstractEndAgent {
    private static final Logger log = LoggerFactory.getLogger(EndAgentImpl.class);
    private final Map taskCache;
    
    public EndAgentImpl(UDPSDK udpSDK, String agentId, String agentName, Map capabilities) {
        super(udpSDK, agentId, agentName, "EndAgent", capabilities);
        this.taskCache = new java.util.concurrent.ConcurrentHashMap<>();
    }
    
    @Override
    protected ResponsePacket createResponse(TaskPacket taskPacket, boolean success, String message) {
        // 缓存任务
        taskCache.put(taskPacket.getTaskId(), taskPacket);
        
        // 创建响应
        ResponseStatus status = success ? ResponseStatus.SUCCESS : ResponseStatus.FAILURE;
        
        return ResponsePacketBuilder.builder()
            .taskId(taskPacket.getTaskId())
            .status(status)
            .message(message)
            .data(processTaskData(taskPacket))
            .timestamp(System.currentTimeMillis())
            .build();
    }
    
    private String processTaskData(TaskPacket taskPacket) {
        // 根据任务类型处理数据
        String taskType = taskPacket.getTaskType();
        Map params = taskPacket.getParams();
        
        switch (taskType) {
            case "skill_invoke":
                return processSkillInvoke(params);
            case "data_collection":
                return processDataCollection(params);
            case "device_control":
                return processDeviceControl(params);
            default:
                return "Unknown task type: " + taskType;
        }
    }
    
    private String processSkillInvoke(Map params) {
        String skillName = (String) params.get("skill");
        Map skillParams = (Map) params.get("params");
        
        // 调用技能逻辑
        log.info("Invoking skill: {} with params: {}", skillName, skillParams);
        
        // 模拟技能执行
        return "Skill " + skillName + " invoked successfully";
    }
    
    private String processDataCollection(Map params) {
        String dataType = (String) params.get("dataType");
        
        // 数据采集逻辑
        log.info("Collecting data of type: {}", dataType);
        
        // 模拟数据采集
        Map collectedData = new java.util.HashMap<>();
        collectedData.put("type", dataType);
        collectedData.put("timestamp", System.currentTimeMillis());
        collectedData.put("value", "Sample data");
        
        return collectedData.toString();
    }
    
    private String processDeviceControl(Map params) {
        String deviceId = (String) params.get("deviceId");
        String command = (String) params.get("command");
        
        // 设备控制逻辑
        log.info("Controlling device {} with command: {}", deviceId, command);
        
        // 模拟设备控制
        return "Device " + deviceId + " controlled with command: " + command;
    }
}

package net.ooder.sdk.agent.impl;

import net.ooder.sdk.agent.RouteAgent;
import net.ooder.sdk.packet.AuthPacket;
import net.ooder.sdk.packet.TaskPacket;
import net.ooder.sdk.packet.RoutePacket;
import net.ooder.sdk.network.udp.SendResult;
import net.ooder.sdk.network.udp.UDPSDK;
import net.ooder.sdk.scene.SceneDefinition;
import net.ooder.sdk.scene.SceneMember;
import net.ooder.sdk.security.EncryptionUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;

public class RouteAgentImpl extends AbstractRouteAgent {
    private static final Logger log = LoggerFactory.getLogger(RouteAgentImpl.class);
    private final Map forwardedTasks;
    
    public RouteAgentImpl(UDPSDK udpSDK, String agentId, String agentName, Map capabilities) {
        super(udpSDK, agentId, agentName, capabilities);
        this.forwardedTasks = new java.util.concurrent.ConcurrentHashMap<>();
    }
    
    @Override
    public CompletableFuture register(String targetMcpId) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                log.info("Registering RouteAgent {} to MCP Agent {}", getAgentId(), targetMcpId);
                
                // 生成认证数据包
                AuthPacket authPacket = AuthPacket.builder()
                    .agentId(getAgentId())
                    .agentName(getAgentName())
                    .agentType(getAgentType())
                    .capabilities(getCapabilities())
                    .timestamp(System.currentTimeMillis())
                    .build();
                
                // 加密认证数据
                String encryptedData = EncryptionUtil.encrypt(authPacket.toString(), getEncryptionKey());
                authPacket.setEncryptedData(encryptedData);
                
                // 发送注册请求
                SendResult result = getUdpSDK().sendAuthRequest(targetMcpId, authPacket);
                
                if (result.isSuccess()) {
                    setMcpAgentId(targetMcpId);
                    setRegistered(true);
                    log.info("RouteAgent {} registered successfully to MCP Agent {}", getAgentId(), targetMcpId);
                    return true;
                } else {
                    log.error("Failed to register RouteAgent {}: {}", getAgentId(), result.getMessage());
                    return false;
                }
            } catch (Exception e) {
                log.error("Error registering RouteAgent: {}", e.getMessage());
                return false;
            }
        }, getExecutorService());
    }
    
    @Override
    public CompletableFuture forwardTask(TaskPacket taskPacket, String endAgentId) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                log.info("Forwarding task {} from EndAgent {} via RouteAgent {}", 
                    taskPacket.getTaskId(), endAgentId, getAgentId());
                
                // 缓存转发的任务
                forwardedTasks.put(taskPacket.getTaskId(), taskPacket);
                
                // 检查EndAgent是否在线
                if (!getEndAgentRoutes().containsKey(endAgentId)) {
                    log.error("EndAgent {} not found in route table", endAgentId);
                    return new SendResult(false, "EndAgent not found");
                }
                
                // 转发任务
                return getUdpSDK().forwardTask(taskPacket, endAgentId);
            } catch (Exception e) {
                log.error("Error forwarding task: {}", e.getMessage());
                return new SendResult(false, e.getMessage());
            }
        }, getExecutorService());
    }
}

package net.ooder.sdk.network.discovery;

import net.ooder.sdk.network.udp.UDPSDK;
import net.ooder.sdk.packet.DiscoveryPacket;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class P2PDiscoveryService {
    private static final Logger log = LoggerFactory.getLogger(P2PDiscoveryService.class);
    private static final int DISCOVERY_PORT = 7777;
    private static final int DISCOVERY_INTERVAL = 5000; // 5秒
    private final UDPSDK udpSDK;
    private final ExecutorService executorService;
    private volatile boolean running;
    
    public P2PDiscoveryService(UDPSDK udpSDK) {
        this.udpSDK = udpSDK;
        this.executorService = Executors.newSingleThreadExecutor();
        this.running = false;
    }
    
    public void start() {
        running = true;
        executorService.submit(this::discoveryLoop);
        log.info("P2P discovery service started");
    }
    
    public void stop() {
        running = false;
        executorService.shutdown();
        log.info("P2P discovery service stopped");
    }
    
    private void discoveryLoop() {
        try (DatagramSocket socket = new DatagramSocket()) {
            socket.setBroadcast(true);
            
            while (running) {
                try {
                    // 发送发现广播
                    sendDiscoveryBroadcast(socket);
                    
                    // 接收响应
                    receiveDiscoveryResponses(socket);
                    
                    // 等待下一次发现
                    Thread.sleep(DISCOVERY_INTERVAL);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                } catch (Exception e) {
                    log.error("Error in discovery loop: {}", e.getMessage());
                }
            }
        } catch (Exception e) {
            log.error("Error starting discovery service: {}", e.getMessage());
        }
    }
    
    private void sendDiscoveryBroadcast(DatagramSocket socket) throws Exception {
        DiscoveryPacket packet = DiscoveryPacket.builder()
            .type("DISCOVERY_REQUEST")
            .senderId(udpSDK.getAgentId())
            .senderType(udpSDK.getAgentType())
            .timestamp(System.currentTimeMillis())
            .build();
        
        byte[] data = packet.toString().getBytes();
        DatagramPacket datagramPacket = new DatagramPacket(
            data, data.length,
            InetAddress.getByName("255.255.255.255"),
            DISCOVERY_PORT
        );
        
        socket.send(datagramPacket);
        log.debug("Sent discovery broadcast");
    }
    
    private void receiveDiscoveryResponses(DatagramSocket socket) throws Exception {
        byte[] buffer = new byte[4096];
        DatagramPacket packet = new DatagramPacket(buffer, buffer.length);
        
        socket.setSoTimeout(1000);
        try {
            socket.receive(packet);
            String response = new String(packet.getData(), 0, packet.getLength());
            log.debug("Received discovery response: {}", response);
            
            // 处理响应
            processDiscoveryResponse(response, packet.getAddress());
        } catch (java.net.SocketTimeoutException e) {
            // 正常超时，继续
        }
    }
    
    private void processDiscoveryResponse(String response, InetAddress address) {
        try {
            DiscoveryPacket packet = DiscoveryPacket.fromString(response);
            if ("DISCOVERY_RESPONSE".equals(packet.getType())) {
                log.info("Discovered peer: {} ({}) at {}", 
                    packet.getSenderId(), packet.getSenderType(), address.getHostAddress());
                
                // 添加到对等节点列表
                udpSDK.addPeer(packet.getSenderId(), address.getHostAddress());
            }
        } catch (Exception e) {
            log.error("Error processing discovery response: {}", e.getMessage());
        }
    }
}