A2A协议最佳实践

本文档总结了A2A（Agent-to-Agent）协议开发的最佳实践。

协议设计最佳实践

1. 消息格式标准化

统一的消息格式

# A2A消息格式
a2a_message = {
    "header": {
        "version": "1.0",
        "message_id": "unique_id",
        "timestamp": "2025-12-19T10:00:00Z",
        "source": "agent_id",
        "destination": "agent_id",
        "message_type": "request|response|notification"
    },
    "body": {
        "action": "action_name",
        "parameters": {},
        "data": {}
    },
    "metadata": {
        "priority": "high|medium|low",
        "ttl": 3600,
        "retry_count": 0
    }
}

2. 错误处理

完善的错误处理机制

class A2AErrorHandler:
    @staticmethod
    def handle_error(error: Exception, context: dict) -> dict:
        """统一错误处理"""
        error_response = {
            "error": True,
            "error_code": type(error).__name__,
            "error_message": str(error),
            "context": context,
            "timestamp": datetime.utcnow().isoformat()
        }
        
        # 根据错误类型处理
        if isinstance(error, TimeoutError):
            error_response["retryable"] = True
        elif isinstance(error, ValueError):
            error_response["retryable"] = False
        
        return error_response

3. 消息路由

高效的路由机制

class A2ARouter:
    def __init__(self):
        self.routes = {}
    
    def register_route(self, pattern: str, handler: callable):
        """注册路由"""
        self.routes[pattern] = handler
    
    def route_message(self, message: dict) -> dict:
        """路由消息"""
        action = message["body"]["action"]
        
        # 查找匹配的路由
        for pattern, handler in self.routes.items():
            if self.match_pattern(pattern, action):
                return handler(message)
        
        # 默认处理
        return self.default_handler(message)

通信最佳实践

1. 异步通信

异步消息处理

import asyncio
from asyncio import Queue

class A2AAgent:
    def __init__(self, agent_id: str):
        self.agent_id = agent_id
        self.message_queue = Queue()
        self.handlers = {}
    
    async def send_message(self, destination: str, message: dict):
        """发送消息"""
        message["header"]["source"] = self.agent_id
        message["header"]["destination"] = destination
        
        # 异步发送
        await self.transport.send(destination, message)
    
    async def receive_message(self):
        """接收消息"""
        message = await self.message_queue.get()
        await self.process_message(message)
    
    async def process_message(self, message: dict):
        """处理消息"""
        action = message["body"]["action"]
        handler = self.handlers.get(action)
        
        if handler:
            await handler(message)
        else:
            await self.default_handler(message)

2. 消息确认

可靠的消息传递

class ReliableMessaging:
    def __init__(self):
        self.pending_messages = {}
        self.ack_timeout = 30
    
    async def send_with_ack(self, destination: str, message: dict):
        """发送带确认的消息"""
        message_id = message["header"]["message_id"]
        self.pending_messages[message_id] = {
            "message": message,
            "timestamp": time.time(),
            "retries": 0
        }
        
        # 发送消息
        await self.send(destination, message)
        
        # 等待确认
        ack = await self.wait_for_ack(message_id, timeout=self.ack_timeout)
        
        if not ack:
            # 重试
            await self.retry_message(message_id)
    
    async def send_ack(self, message_id: str):
        """发送确认"""
        ack_message = {
            "header": {
                "message_type": "ack",
                "original_message_id": message_id
            }
        }
        await self.send(ack_message)

3. 消息队列

消息队列管理

from queue import PriorityQueue

class MessageQueue:
    def __init__(self):
        self.queue = PriorityQueue()
    
    def enqueue(self, message: dict, priority: int = 0):
        """入队"""
        self.queue.put((priority, time.time(), message))
    
    def dequeue(self):
        """出队"""
        if not self.queue.empty():
            return self.queue.get()[2]
        return None
    
    def size(self):
        """队列大小"""
        return self.queue.qsize()

安全最佳实践

1. 身份验证

Agent身份验证

import jwt
import hashlib

class A2AAuthentication:
    def __init__(self, secret_key: str):
        self.secret_key = secret_key
    
    def generate_token(self, agent_id: str) -> str:
        """生成认证令牌"""
        payload = {
            "agent_id": agent_id,
            "exp": datetime.utcnow() + timedelta(hours=24)
        }
        return jwt.encode(payload, self.secret_key, algorithm="HS256")
    
    def verify_token(self, token: str) -> dict:
        """验证令牌"""
        try:
            payload = jwt.decode(token, self.secret_key, algorithms=["HS256"])
            return payload
        except jwt.ExpiredSignatureError:
            raise ValueError("Token expired")
        except jwt.InvalidTokenError:
            raise ValueError("Invalid token")

2. 消息加密

端到端加密

from cryptography.fernet import Fernet

class MessageEncryption:
    def __init__(self, key: bytes):
        self.cipher = Fernet(key)
    
    def encrypt_message(self, message: dict) -> str:
        """加密消息"""
        message_str = json.dumps(message)
        encrypted = self.cipher.encrypt(message_str.encode())
        return encrypted.decode()
    
    def decrypt_message(self, encrypted: str) -> dict:
        """解密消息"""
        decrypted = self.cipher.decrypt(encrypted.encode())
        return json.loads(decrypted.decode())

监控最佳实践

1. 消息追踪

完整的追踪机制

class MessageTracker:
    def __init__(self):
        self.traces = {}
    
    def track_message(self, message_id: str, event: str, data: dict):
        """追踪消息"""
        if message_id not in self.traces:
            self.traces[message_id] = []
        
        self.traces[message_id].append({
            "event": event,
            "timestamp": datetime.utcnow().isoformat(),
            "data": data
        })
    
    def get_trace(self, message_id: str) -> list:
        """获取追踪信息"""
        return self.traces.get(message_id, [])

总结

遵循这些最佳实践可以：

1. 提高可靠性：通过错误处理和消息确认
2. 优化性能：通过异步通信和队列管理
3. 增强安全性：通过身份验证和加密
4. 改善可维护性：通过标准化和监控