第八章：最佳实践

错误处理

重试机制

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

class RobustA2AClient(A2AClient):
    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10))
    async def send_task_with_retry(self, task_id: str, message: str):
        return await self.send_task(task_id, message)

超时处理

import asyncio

async def send_task_with_timeout(client: A2AClient, task_id: str, message: str, timeout: float = 30.0):
    try:
        return await asyncio.wait_for(
            client.send_task(task_id, message),
            timeout=timeout
        )
    except asyncio.TimeoutError:
        return {"error": "Task timeout"}

监控

健康检查

async def health_check(client: A2AClient) -> bool:
    try:
        async with httpx.AsyncClient() as http_client:
            response = await http_client.get(
                f"{client.base_url}/.well-known/agent.json",
                timeout=5.0
            )
            return response.status_code == 200
    except:
        return False

日志记录

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("a2a")

async def send_task_with_logging(client: A2AClient, task_id: str, message: str):
    logger.info(f"Sending task {task_id} to {client.base_url}")

    try:
        result = await client.send_task(task_id, message)
        logger.info(f"Task {task_id} completed: {result}")
        return result
    except Exception as e:
        logger.error(f"Task {task_id} failed: {e}")
        raise

安全建议

安全最佳实践：
1. 使用 HTTPS
2. 验证所有输入
3. 实施速率限制
4. 使用 API Key 或 OAuth
5. 记录所有操作日志
6. 定期审计访问权限

小结

最佳实践要点：

• 错误处理：重试机制、超时处理
• 监控：健康检查、日志记录
• 安全：HTTPS、认证、审计

完成本教程后，你应该能够使用 A2A 协议构建多智能体协作系统。