【Agent实战】深度实战！A2A 代理构建最佳实践全解析，踩坑经验一文打尽

class AgentExecutor:    """Bridge between A2A protocol and agent implementation"""    def __init__(self, agent_class):        self.agent = agent_class()        self.supported_content_types = ['text', 'text/plain']    async def execute(self, context: ExecutionContext, event_queue: EventQueue):        try:            user_input = context.get_user_input()           if context.is_streaming():               await self._handle_streaming(user_input, context, event_queue)           else:               await self._handle_non_streaming(user_input, context, event_queue)       except Exception as e:           await self._handle_error(e, context, event_queue)

from pydantic import BaseModelfrom typing import Literalclass ResponseFormat(BaseModel):    """Standardized response format for all agents"""    status: Literal['input_required', 'completed', 'error']    message: str    metadata: Optional[Dict[str, Any]] = Noneclass CurrencyAgent:    def __init__(self):        self.graph = create_react_agent(            self.model,            tools=self.tools,            response_format=ResponseFormat,        )

class MathAgent:    async def _init_mcp_tools(self):        try:            server_params = StdioServerParameters(                command="python",                args=[str(self.math_server_path)],            )           async with stdio_client(server_params) as (read, write):                session = ClientSession(read, write)               await session.initialize()                tools = await load_mcp_tools(session)               return tools       except Exception as e:           print(f"MCP failed: {e}, falling back to direct API")           return self._create_direct_tools()

class MathAgent:    async def process_request(self, request: str) -> str:        server_params = StdioServerParameters(            command="python",            args=[str(self.math_server_path)],        )       async with stdio_client(server_params) as (read, write):           async with ClientSession(read, write) as session:               await session.initialize()                tools = await load_mcp_tools(session)                agent = create_react_agent(self.model, tools)               return await agent.ainvoke({"messages": request})

class SmartOrchestrator:    def _calculate_agent_score(self, request: str, agent_card: AgentCard) -> tuple[float, List[str]]:        score = 0.0        matched_skills = []        request_lower = request.lower()       # Keyword matching (weight: 1.0)        keywords = [tag for skill in agent_card.skills for tag in (skill.tags or [])]       for keyword in keywords:           if keyword.lower() in request_lower:                score += 1.0       # Skill matching (weight: 1.5)       for skill in agent_card.skills:           if self._skill_matches_request(skill.name, request):                score += 1.5                matched_skills.append(skill.name)       return score, matched_skills    async def _analyze_request(self, state: RouterState) -> RouterState:        request = state["request"]        best_agent = None        best_score = 0.0        agent_scores = {}       for agent_id, agent_card in self.agents.items():            score, matched_skills = self._calculate_agent_score(request, agent_card)            agent_scores[agent_id] = score           if score > best_score:                best_score = score                best_agent = agent_id        confidence = min(best_score / 5.0, 1.0)        state.update({           "selected_agent": best_agent,           "confidence": confidence,           "metadata": {"agent_scores": agent_scores}        })       return state

async def display_available_agents(httpx_client, agent_url: str, card):   if "orchestrator" in card.name.lower():        agent_endpoints = [            ("ArgoCD Agent", "http://localhost:8001"),            ("Currency Agent", "http://localhost:8002"),            ("Math Agent", "http://localhost:8003")        ]        available_agents = []       for name, endpoint in agent_endpoints:           try:                response = await httpx_client.get(                   f"{endpoint}/.well-known/agent.json",                    timeout=2.0                )               if response.status_code == 200:                    agent_data = response.json()                    available_agents.append({                       "name": agent_data.get("name", name),                       "description": agent_data.get("description", ""),                       "url": endpoint,                       "skills": agent_data.get("skills", [])                    })           except:               pass       if available_agents:           print(f"Found {len(available_agents)} available agents:")           for i, agent in enumerate(available_agents, 1):               print(f"{i}. {agent['name']} ({agent['url']})")               print(f"   Description: {agent['description']}")               if agent['skills']:                    skills = [skill.get('name', 'Unknown') for skill in agent['skills'][:3]]                   print(f"   Skills: {', '.join(skills)}")

def validate_mathematical_expression(expression: str) -> bool:   if len(expression) > 1000:       return False    dangerous_patterns = ['import', 'exec', 'eval', '__', 'subprocess']   if any(pattern in expression.lower() for pattern in dangerous_patterns):       return False   import re    allowed_chars = re.compile(r'^[a-zA-Z0-9+\-*/()^.,\s=<>!]+$')   return bool(allowed_chars.match(expression))def validate_currency_code(currency: str) -> bool:   return len(currency) == 3 and currency.isalpha() and currency.isupper()def sanitize_user_input(user_input: str) -> str:   import re    sanitized = re.sub(r'[<>"\']', '', user_input)   return sanitized.strip()[:10000]

import resourceimport asynciofrom contextlib import asynccontextmanager@asynccontextmanagerasync def resource_limits(cpu_seconds=30, memory_mb=512):   try:        resource.setrlimit(resource.RLIMIT_CPU, (cpu_seconds, cpu_seconds))        memory_bytes = memory_mb * 1024 * 1024        resource.setrlimit(resource.RLIMIT_AS, (memory_bytes, memory_bytes))       yield   finally:        resource.setrlimit(resource.RLIMIT_CPU, (resource.RLIM_INFINITY, resource.RLIM_INFINITY))        resource.setrlimit(resource.RLIMIT_AS, (resource.RLIM_INFINITY, resource.RLIM_INFINITY))class MathAgent:   async def process_request(self, request: str) -> str:       async with resource_limits(cpu_seconds=30, memory_mb=512):           return await self._unsafe_process_request(request)

from enum import Enumimport timeclass CircuitState(Enum):    CLOSED = "closed"    OPEN = "open"    HALF_OPEN = "half_open"class CircuitBreaker:   def __init__(self, failure_threshold=5, recovery_timeout=60):        self.failure_threshold = failure_threshold        self.recovery_timeout = recovery_timeout        self.failure_count = 0        self.last_failure_time = None        self.state = CircuitState.CLOSED   async def call(self, func, *args, **kwargs):       if self.state == CircuitState.OPEN:           if time.time() - self.last_failure_time > self.recovery_timeout:                self.state = CircuitState.HALF_OPEN           else:               raise Exception("Circuit breaker is OPEN")       try:            result = await func(*args, **kwargs)            self._on_success()           return result       except Exception as e:            self._on_failure()           raise e   def _on_success(self):        self.failure_count = 0        self.state = CircuitState.CLOSED   def _on_failure(self):        self.failure_count += 1        self.last_failure_time = time.time()       if self.failure_count >= self.failure_threshold:            self.state = CircuitState.OPENclass CurrencyAgent:   def __init__(self):        self.api_circuit_breaker = CircuitBreaker()   async def get_exchange_rate(self, from_curr, to_curr):       try:           return await self.api_circuit_breaker.call(                self._call_frankfurter_api, from_curr, to_curr            )       except Exception:           return self._get_cached_rate(from_curr, to_curr)

import httpxfrom typing import Optionalclass HTTPClientManager:   """Singleton HTTP client manager with connection pooling"""    _instance: Optional['HTTPClientManager'] = None    _client: Optional[httpx.AsyncClient] = None   def __new__(cls):       if cls._instance is None:            cls._instance = super().__new__(cls)       return cls._instance   async def get_client(self) -> httpx.AsyncClient:       if self._client is None:            self._client = httpx.AsyncClient(                timeout=30.0,                limits=httpx.Limits(                    max_connections=100,                    max_keepalive_connections=20                ),                headers={                   "User-Agent": "A2A-Agent/1.0",                   "Accept": "application/json"                }            )       return self._client   async def close(self):       if self._client:           await self._client.aclose()            self._client = Noneclass CurrencyAgent:   async def get_exchange_rate(self, currency_from: str, currency_to: str):        client_manager = HTTPClientManager()        client = await client_manager.get_client()        response = await client.get(           f'https://api.frankfurter.app/latest',            params={'from': currency_from, 'to': currency_to}        )       return response.json()

from functools import lru_cacheimport timefrom typing import Dict, Tuple, Anyimport asyncioclass TTLCache:   """Time-based cache with TTL support"""   def __init__(self, ttl_seconds: int = 300):        self.ttl_seconds = ttl_seconds        self.cache: Dict[str, Tuple[Any, float]] = {}        self.lock = asyncio.Lock()   async def get(self, key: str) -> Any:       async with self.lock:           if key in self.cache:                value, timestamp = self.cache[key]               if time.time() - timestamp < self.ttl_seconds:                   return value               else:                   del self.cache[key]           return None   async def set(self, key: str, value: Any):       async with self.lock:            self.cache[key] = (value, time.time())class CurrencyAgent:   def __init__(self):        self.rate_cache = TTLCache(ttl_seconds=300)   async def get_exchange_rate(self, currency_from: str, currency_to: str):        cache_key = f"{currency_from}_{currency_to}"       if await self.rate_cache.get(cache_key):           return await self.rate_cache.get(cache_key)        result = await self._fetch_from_api(currency_from, currency_to)       await self.rate_cache.set(cache_key, result)       return result