Agent-to-Agent Protocol: A2A for Multi-Agent Systems in 2026

# Agent Discovery Example
class AgentDiscovery:
    def __init__(self, discovery_service_url: str):
        self.discovery_service = discovery_service_url
    
    async def discover_agents(self, capability: str) -> List[Dict]:
        """Discover agents with specific capabilities"""
        response = await self._query_discovery_service({
            "query": {
                "capabilities": capability,
                "version": ">=1.0.0"
            }
        })
        return response.get("agents", [])
    
    async def register_agent(self, agent_card: Dict) -> bool:
        """Register agent with discovery service"""
        response = await self._post_to_discovery_service({
            "action": "register",
            "agent": agent_card
        })
        return response.get("success", False)

# Task Delegation Example
class TaskDelegator:
    def __init__(self, discovery_service: AgentDiscovery):
        self.discovery = discovery_service
    
    async delegate_task(self, task_description: str, 
                       required_capabilities: List[str]) -> str:
        """Delegate task to appropriate agent"""
        
        # Find agents with required capabilities
        agents = []
        for capability in required_capabilities:
            capable_agents = await self.discovery.discover_agents(capability)
            agents.extend(capable_agents)
        
        # Select best agent based on criteria
        selected_agent = self._select_best_agent(agents, task_description)
        
        # Delegate task
        task_id = await self._delegate_to_agent(
            selected_agent, 
            task_description
        )
        
        return task_id

# Stateful Conversation Example
class AgentConversation:
    def __init__(self, agent_url: str):
        self.agent_url = agent_url
        self.conversation_id = None
        self.messages = []
    
    async def start_conversation(self, initial_message: str) -> str:
        """Start a new conversation with agent"""
        response = await self._send_request({
            "method": "conversation/start",
            "params": {
                "message": initial_message
            }
        })
        self.conversation_id = response["conversationId"]
        self.messages.append(initial_message)
        return response["response"]
    
    async def continue_conversation(self, message: str) -> str:
        """Continue existing conversation"""
        response = await self._send_request({
            "method": "conversation/continue",
            "params": {
                "conversationId": self.conversation_id,
                "message": message
            }
        })
        self.messages.append(message)
        return response["response"]

# Real-Time Collaboration Example
class RealTimeCollaboration:
    def __init__(self, agent_urls: List[str]):
        self.agents = agent_urls
        self.collaboration_id = None
    
    async def start_collaboration(self, topic: str) -> str:
        """Start real-time collaboration session"""
        response = await self._send_to_all_agents({
            "method": "collaboration/start",
            "params": {
                "topic": topic,
                "participants": self.agents
            }
        })
        self.collaboration_id = response["collaborationId"]
        return self.collaboration_id
    
    async def share_insight(self, insight: str) -> None:
        """Share insight with all collaborating agents"""
        await self._send_to_all_agents({
            "method": "collaboration/share",
            "params": {
                "collaborationId": self.collaboration_id,
                "insight": insight
            }
        })

{
  "name": "Research Agent v2.1",
  "description": "Advanced research agent capable of web search, academic paper analysis, and trend identification",
  "url": "https://research-agent.example.com/a2a",
  "version": "2.1.0",
  "protocolVersion": "1.0.0",
  "provider": {
    "name": "ResearchAI Inc.",
    "url": "https://researchai.example.com"
  },
  "capabilities": {
    "streaming": true,
    "pushNotifications": true,
    "realTimeUpdates": true,
    "batchProcessing": true,
    "maxConcurrentTasks": 10,
    "rateLimit": {
      "requestsPerMinute": 100,
      "burstLimit": 20
    }
  },
  "skills": [
    {
      "id": "web-search-advanced",
      "name": "Advanced Web Search",
      "description": "Comprehensive web search with source verification",
      "inputModes": ["text", "url"],
      "outputModes": ["text", "structured"],
      "parameters": {
        "maxResults": 50,
        "timeRange": ["1d", "7d", "30d", "1y"],
        "sources": ["web", "academic", "news"]
      }
    },
    {
      "id": "academic-analysis",
      "name": "Academic Paper Analysis",
      "description": "Analyze academic papers and extract key insights",
      "inputModes": ["pdf", "text", "url"],
      "outputModes": ["summary", "key_points", "citations"],
      "parameters": {
        "maxPages": 100,
        "citationStyle": "APA"
      }
    },
    {
      "id": "trend-identification",
      "name": "Trend Identification",
      "description": "Identify emerging trends from data sources",
      "inputModes": ["text", "csv", "json"],
      "outputModes": ["trends", "visualization", "report"],
      "parameters": {
        "timeframe": "auto",
        "confidenceThreshold": 0.8
      }
    }
  ],
  "authentication": {
    "type": "oauth2",
    "flows": ["authorization_code", "client_credentials"],
    "scopes": ["read", "write", "admin"],
    "tokenUrl": "https://auth.example.com/oauth/token",
    "authorizationUrl": "https://auth.example.com/oauth/authorize"
  },
  "pricing": {
    "model": "per_task",
    "tiers": [
      {
        "name": "Free",
        "maxTasksPerDay": 100,
        "features": ["basic_search", "simple_analysis"]
      },
      {
        "name": "Pro",
        "maxTasksPerDay": 1000,
        "features": ["advanced_search", "academic_analysis", "trend_identification"]
      }
    ]
  },
  "metadata": {
    "createdAt": "2026-01-15T10:30:00Z",
    "updatedAt": "2026-03-01T14:20:00Z",
    "uptime": 99.8,
    "responseTime": "150ms",
    "supportedLanguages": ["en", "es", "fr", "de", "zh"]
  }
}

# Agent Discovery Implementation
class AgentDiscoveryService:
    def __init__(self):
        self.agent_registry = {}
    
    async def register_agent(self, agent_card: Dict) -> str:
        """Register agent and return agent ID"""
        agent_id = self._generate_agent_id(agent_card)
        self.agent_registry[agent_id] = {
            "card": agent_card,
            "registeredAt": datetime.utcnow(),
            "lastSeen": datetime.utcnow(),
            "status": "online"
        }
        return agent_id
    
    async def discover_agents(self, filters: Dict) -> List[Dict]:
        """Discover agents matching filters"""
        matching_agents = []
        
        for agent_id, agent_data in self.agent_registry.items():
            if self._matches_filters(agent_data["card"], filters):
                matching_agents.append({
                    "agentId": agent_id,
                    "card": agent_data["card"],
                    "status": agent_data["status"],
                    "responseTime": agent_data.get("avgResponseTime", 0)
                })
        
        return matching_agents
    
    async def get_agent_capabilities(self, agent_id: str) -> Dict:
        """Get detailed capabilities for specific agent"""
        agent_data = self.agent_registry.get(agent_id)
        if not agent_data:
            raise ValueError(f"Agent not found: {agent_id}")
        
        return {
            "skills": agent_data["card"]["skills"],
            "capabilities": agent_data["card"]["capabilities"],
            "availability": self._calculate_availability(agent_data)
        }

# Python 3.9+ with required packages
pip install aiohttp httpx pydantic python-jose cryptography

# For production deployments
pip install redis aioredis uvicorn fastapi

# Development tools
pip install pytest pytest-asyncio black isort mypy

#!/usr/bin/env python3
"""A2A Protocol Client Implementation."""

import asyncio
import json
import logging
from typing import Any, Dict, List, Optional, Union
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
import aiohttp
from aiohttp import ClientSession, ClientTimeout
from pydantic import BaseModel, Field, validator

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class MessageRole(Enum):
    """Message roles in A2A conversations."""
    USER = "user"
    AGENT = "agent"
    SYSTEM = "system"
    TOOL = "tool"

class TaskStatus(Enum):
    """Task status enumeration."""
    SUBMITTED = "submitted"
    WORKING = "working"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"

@dataclass
class A2AMessage:
    """A2A message structure."""
    role: MessageRole
    content: str
    metadata: Optional[Dict[str, Any]] = None
    timestamp: datetime = field(default_factory=datetime.utcnow)
    message_id: str = field(default_factory=lambda: str(uuid.uuid4()))

@dataclass
class Task:
    """A2A task structure."""
    id: str
    status: TaskStatus
    messages: List[A2AMessage] = field(default_factory=list)
    result: Optional[Any] = None
    created_at: datetime = field(default_factory=datetime.utcnow)
    updated_at: datetime = field(default_factory=datetime.utcnow)
    metadata: Dict[str, Any] = field(default_factory=dict)

class AgentCard(BaseModel):
    """Agent Card model for capability discovery."""
    name: str
    description: str
    url: str
    version: str = "1.0.0"
    protocol_version: str = Field(default="1.0.0", alias="protocolVersion")
    
    capabilities: Dict[str, Any] = Field(default_factory=dict)
    skills: List[Dict[str, Any]] = Field(default_factory=list)
    
    authentication: Optional[Dict[str, Any]] = None
    pricing: Optional[Dict[str, Any]] = None
    metadata: Dict[str, Any] = Field(default_factory=dict)
    
    class Config:
        allow_population_by_field_name = True

class A2AClient:
    """Production-ready A2A client with retry, timeout, and error handling."""
    
    def __init__(
        self,
        agent_url: str,
        auth_token: Optional[str] = None,
        timeout: int = 30,
        max_retries: int = 3,
        session: Optional[ClientSession] = None
    ):
        self.agent_url = agent_url.rstrip('/')
        self.auth_token = auth_token
        self.timeout = timeout
        self.max_retries = max_retries
        self._session = session
        self._request_id = 0
        
        # Connection pool configuration
        self._connector = aiohttp.TCPConnector(
            limit=100,
            limit_per_host=20,
            ttl_dns_cache=300
        )
    
    async def __aenter__(self):
        """Async context manager entry."""
        if not self._session:
            self._session = aiohttp.ClientSession(
                connector=self._connector,
                timeout=ClientTimeout(total=self.timeout)
            )
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """Async context manager exit."""
        if self._session:
            await self._session.close()
    
    async def get_agent_card(self) -> AgentCard:
        """Fetch agent capabilities with validation."""
        try:
            response = await self._request_with_retry(
                "agent/getAgentCard",
                {}
            )
            return AgentCard(**response)
        except Exception as e:
            logger.error(f"Failed to get agent card: {e}")
            raise
    
    async def create_task(
        self,
        message: str,
        skills: Optional[List[str]] = None,
        metadata: Optional[Dict[str, Any]] = None
    ) -> Task:
        """Create a new task with comprehensive error handling."""
        
        params = {
            "message": {
                "role": "user",
                "content": message
            }
        }
        
        if skills:
            params["skills"] = skills
        
        if metadata:
            params["metadata"] = metadata
        
        try:
            response = await self._request_with_retry("tasks/create", params)
            
            return Task(
                id=response["id"],
                status=TaskStatus(response["status"]),
                metadata=response.get("metadata", {})
            )
        except Exception as e:
            logger.error(f"Failed to create task: {e}")
            raise
    
    async def get_task_result(self, task_id: str) -> Task:
        """Get task status and result with polling support."""
        
        response = await self._request_with_retry(
            "tasks/get",
            {"id": task_id}
        )
        
        messages = [
            A2AMessage(
                role=MessageRole(m["role"]),
                content=m["content"],
                metadata=m.get("metadata")
            )
            for m in response.get("messages", [])
        ]
        
        return Task(
            id=response["id"],
            status=TaskStatus(response["status"]),
            messages=messages,
            result=response.get("result"),
            metadata=response.get("metadata", {})
        )
    
    async def poll_task_result(
        self,
        task_id: str,
        interval: float = 1.0,
        timeout: float = 300.0
    ) -> Task:
        """Poll for task completion with timeout."""
        start_time = asyncio.get_event_loop().time()
        
        while True:
            task = await self.get_task_result(task_id)
            
            if task.status in [TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.CANCELLED]:
                return task
            
            elapsed = asyncio.get_event_loop().time() - start_time
            if elapsed > timeout:
                raise TimeoutError(f"Task {task_id} timed out after {timeout} seconds")
            
            await asyncio.sleep(interval)
    
    async def send_message(self, task_id: str, message: str) -> Task:
        """Send follow-up message to existing task."""
        
        response = await self._request_with_retry(
            "tasks/sendMessage",
            {
                "id": task_id,
                "message": {
                    "role": "user",
                    "content": message
                }
            }
        )
        
        messages = [
            A2AMessage(
                role=MessageRole(m["role"]),
                content=m["content"],
                metadata=m.get("metadata")
            )
            for m in response.get("messages", [])
        ]
        
        return Task(
            id=response["id"],
            status=TaskStatus(response["status"]),
            messages=messages,
            metadata=response.get("metadata", {})
        )
    
    async def cancel_task(self, task_id: str) -> bool:
        """Cancel a running task."""
        response = await self._request_with_retry(
            "tasks/cancel",
            {"id": task_id}
        )
        return response.get("success", False)
    
    async def _request_with_retry(
        self,
        method: str,
        params: Dict[str, Any],
        retry_count: int = 0
    ) -> Dict[str, Any]:
        """Make JSON-RPC request with exponential backoff retry."""
        
        self._request_id += 1
        payload = {
            "jsonrpc": "2.0",
            "id": self._request_id,
            "method": method,
            "params": params
        }
        
        headers = {
            "Content-Type": "application/json",
            "User-Agent": "A2A-Client/1.0.0"
        }
        
        if self.auth_token:
            headers["Authorization"] = f"Bearer {self.auth_token}"
        
        try:
            async with self._session.post(
                f"{self.agent_url}/a2a",
                json=payload,
                headers=headers
            ) as response:
                if response.status != 200:
                    raise aiohttp.ClientError(
                        f"HTTP {response.status}: {await response.text()}"
                    )
                
                result = await response.json()
                
                # Check for JSON-RPC error
                if "error" in result:
                    error = result["error"]
                    raise Exception(
                        f"JSON-RPC error {error.get('code')}: {error.get('message')}"
                    )
                
                return result.get("result", {})
                
        except Exception as e:
            if retry_count < self.max_retries:
                wait_time = 2 ** retry_count  # Exponential backoff
                logger.warning(
                    f"Request failed, retrying in {wait_time}s: {e}"
                )
                await asyncio.sleep(wait_time)
                return await self._request_with_retry(
                    method, params, retry_count + 1
                )
            else:
                logger.error(f"Request failed after {self.max_retries} retries: {e}")
                raise
    
    async def health_check(self) -> bool:
        """Check if agent is healthy and responding."""
        try:
            await self._session.get(f"{self.agent_url}/health", timeout=5)
            return True
        except Exception:
            return False
    
    async def get_agent_metrics(self) -> Dict[str, Any]:
        """Get agent performance metrics."""
        response = await self._request_with_retry("agent/getMetrics", {})
        return response

#!/usr/bin/env python3
"""A2A Protocol Server Implementation."""

import asyncio
import json
import logging
import uuid
from typing import Any, Dict, List, Optional, Callable
from datetime import datetime
from aiohttp import web
from aiohttp.web import Request, Response
import redis.asyncio as redis

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class A2AServer:
    """Production-ready A2A server with task management and persistence."""
    
    def __init__(
        self,
        agent_name: str,
        agent_description: str,
        redis_url: Optional[str] = None,
        max_concurrent_tasks: int = 100
    ):
        self.agent_name = agent_name
        self.agent_description = agent_description
        self.max_concurrent_tasks = max_concurrent_tasks
        
        # Task management
        self.tasks: Dict[str, Dict] = {}
        self.task_queue = asyncio.Queue(maxsize=max_concurrent_tasks)
        
        # Redis for persistence (optional)
        self.redis_client = None
        if redis_url:
            self.redis_client = redis.from_url(redis_url)
        
        # Worker pool
        self.workers = []
        self.is_running = False
        
        # Register JSON-RPC methods
        self.methods = {
            "agent/getAgentCard": self.get_agent_card,
            "agent/getMetrics": self.get_agent_metrics,
            "tasks/create": self.create_task,
            "tasks/get": self.get_task,
            "tasks/sendMessage": self.send_message,
            "tasks/cancel": self.cancel_task,
            "tasks/list": self.list_tasks,
            "health": self.health_check
        }
    
    async def start(self, host: str = "0.0.0.0", port: int = 8080):
        """Start the A2A server."""
        app = web.Application()
        app.router.add_post("/a2a", self.handle_request)
        app.router.add_get("/health", self.health_endpoint)
        app.router.add_get("/", self.root_endpoint)
        
        # Start worker pool
        self.is_running = True
        for i in range(5):  # 5 worker threads
            worker = asyncio.create_task(self._worker(i))
            self.workers.append(worker)
        
        logger.info(f"Starting A2A server on {host}:{port}")
        runner = web.AppRunner(app)
        await runner.setup()
        site = web.TCPSite(runner, host, port)
        await site.start()
        
        return runner
    
    async def stop(self):
        """Stop the A2A server gracefully."""
        self.is_running = False
        
        # Wait for workers to finish
        for worker in self.workers:
            worker.cancel()
        
        # Close Redis connection
        if self.redis_client:
            await self.redis_client.close()
        
        logger.info("A2A server stopped")
    
    async def handle_request(self, request: Request) -> Response:
        """Handle incoming JSON-RPC requests."""
        try:
            payload = await request.json()
            
            method = payload.get("method")
            params = payload.get("params", {})
            request_id = payload.get("id", 1)
            
            if method not in self.methods:
                return self._error_response(
                    request_id,
                    -32601,
                    f"Method not found: {method}"
                )
            
            handler = self.methods[method]
            result = await handler(params)
            
            return self._success_response(request_id, result)
            
        except json.JSONDecodeError:
            return self._error_response(
                1, -32700, "Parse error: Invalid JSON"
            )
        except Exception as e:
            logger.error(f"Request handling error: {e}")
            return self._error_response(
                payload.get("id", 1) if 'payload' in locals() else 1,
                -32000,
                f"Server error: {str(e)}"
            )
    
    async def get_agent_card(self, params: Dict) -> Dict:
        """Return agent capabilities."""
        return {
            "name": self.agent_name,
            "description": self.agent_description,
            "url": "https://agent.example.com/a2a",
            "version": "1.0.0",
            "protocolVersion": "1.0.0",
            "capabilities": {
                "streaming": True,
                "pushNotifications": True,
                "realTimeUpdates": True,
                "maxConcurrentTasks": self.max_concurrent_tasks,
                "rateLimit": {
                    "requestsPerMinute": 1000,
                    "burstLimit": 100
                }
            },
            "skills": [
                {
                    "id": "processing",
                    "name": "Data Processing",
                    "description": "Process and analyze data",
                    "inputModes": ["text", "json", "csv"],
                    "outputModes": ["text", "json", "structured"]
                }
            ],
            "authentication": {
                "type": "bearer",
                "scopes": ["read", "write"]
            },
            "metadata": {
                "uptime": 99.9,
                "responseTime": "50ms",
                "supportedLanguages": ["en"]
            }
        }
    
    async def create_task(self, params: Dict) -> Dict:
        """Create new task and add to queue."""
        task_id = str(uuid.uuid4())
        message = params["message"]
        
        task = {
            "id": task_id,
            "status": "submitted",
            "messages": [message],
            "createdAt": datetime.utcnow().isoformat() + "Z",
            "updatedAt": datetime.utcnow().isoformat() + "Z",
            "metadata": params.get("metadata", {})
        }
        
        # Store task
        self.tasks[task_id] = task
        
        # Persist to Redis if available
        if self.redis_client:
            await self.redis_client.set(
                f"task:{task_id}",
                json.dumps(task)
            )
        
        # Add to processing queue
        await self.task_queue.put(task_id)
        
        return {"id": task_id, "status": "submitted"}
    
    async def get_task(self, params: Dict) -> Dict:
        """Get task status and result."""
        task_id = params["id"]
        
        # Try to get from memory first
        task = self.tasks.get(task_id)
        
        # If not in memory, try Redis
        if not task and self.redis_client:
            task_data = await self.redis_client.get(f"task:{task_id}")
            if task_data:
                task = json.loads(task_data)
        
        if not task:
            raise ValueError(f"Task not found: {task_id}")
        
        return task
    
    async def _worker(self, worker_id: int):
        """Worker process for handling tasks."""
        logger.info(f"Worker {worker_id} started")
        
        while self.is_running:
            try:
                task_id = await self.task_queue.get()
                
                # Process task
                await self._process_task(task_id)
                
                self.task_queue.task_done()
                
            except asyncio.CancelledError:
                break
            except Exception as e:
                logger.error(f"Worker {worker_id} error: {e}")
    
    async def _process_task(self, task_id: str):
        """Process a task."""
        task = self.tasks[task_id]
        
        # Update status
        task["status"] = "working"
        task["updatedAt"] = datetime.utcnow().isoformat() + "Z"
        
        try:
            # Process the task (implement your agent logic here)
            result = await self._agent_process(task["messages"][0]["content"])
            
            # Store result
            task["status"] = "completed"
            task["result"] = result
            task["updatedAt"] = datetime.utcnow().isoformat() + "Z"
            
        except Exception as e:
            task["status"] = "failed"
            task["error"] = str(e)
            task["updatedAt"] = datetime.utcnow().isoformat() + "Z"
        
        # Update Redis
        if self.redis_client:
            await self.redis_client.set(
                f"task:{task_id}",
                json.dumps(task)
            )
    
    async def _agent_process(self, content: str) -> Any:
        """Implement your agent's processing logic here."""
        # This is where your agent's intelligence goes
        return f"Processed: {content}"
    
    def _success_response(self, request_id: int, result: Any) -> Response:
        return web.json_response({
            "jsonrpc": "2.0",
            "id": request_id,
            "result": result
        })
    
    def _error_response(self, request_id: int, 
                       code: int, message: str) -> Response:
        return web.json_response({
            "jsonrpc": "2.0",
            "id": request_id,
            "error": {
                "code": code,
                "message": message
            }
        })
    
    async def health_endpoint(self, request: Request) -> Response:
        """Health check endpoint."""
        return web.json_response({
            "status": "healthy",
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "metrics": {
                "activeTasks": len([t for t in self.tasks.values() 
                                  if t["status"] in ["submitted", "working"]]),
                "queueSize": self.task_queue.qsize(),
                "totalTasks": len(self.tasks)
            }
        })
    
    async def root_endpoint(self, request: Request) -> Response:
        """Root endpoint with documentation."""
        return web.Response(
            text="A2A Protocol Server - Use /a2a for JSON-RPC requests"
        )

# Example: Research and Write workflow with A2A

class ResearchWorkflow:
    """Multi-agent workflow using A2A"""
    
    def __init__(self):
        self.research_agent = A2AClient(
            "https://research-agent.example.com/a2a"
        )
        self.writer_agent = A2AClient(
            "https://writer-agent.example.com/a2a"
        )
        self.critic_agent = A2AClient(
            "https://critic-agent.example.com/a2a"
        )
    
    async def execute(self, topic: str) -> Dict:
        """Execute research and writing workflow"""
        
        # Step 1: Research
        print("🔍 Step 1: Gathering research...")
        research_task = await self.research_agent.create_task(
            f"Research {topic} thoroughly. "
            "Find key concepts, latest developments, and expert opinions.",
            skills=["web-search", "summarize"]
        )
        
        # Wait for research to complete
        while True:
            result = await self.research_agent.get_task_result(
                research_task.id
            )
            if result.status == "completed":
                break
            await asyncio.sleep(1)
        
        research_summary = result.result
        print(f"✅ Research complete: {len(research_summary)} chars")
        
        # Step 2: Write article
        print("✍️ Step 2: Writing article...")
        write_task = await self.writer_agent.create_task(
            f"Write a comprehensive article about {topic}. "
            f"Use this research: {research_summary}",
            skills=["article-writing"]
        )
        
        while True:
            result = await self.writer_agent.get_task_result(
                write_task.id
            )
            if result.status == "completed":
                break
            await asyncio.sleep(1)
        
        article = result.result
        print(f"✅ Article complete: {len(article)} chars")
        
        # Step 3: Review
        print("📝 Step 3: Getting review...")
        review_task = await self.critic_agent.create_task(
            f"Review this article for accuracy and clarity: {article}",
            skills=["critique"]
        )
        
        while True:
            result = await self.critic_agent.get_task_result(
                review_task.id
            )
            if result.status == "completed":
                break
            await asyncio.sleep(1)
        
        feedback = result.result
        print(f"✅ Review complete")
        
        return {
            "article": article,
            "feedback": feedback,
            "research": research_summary
        }

# Secure A2A Implementation
class SecureA2AServer:
    def __init__(self):
        self.rate_limiter = RateLimiter(requests_per_minute=1000)
        self.auth_validator = AuthValidator()
        self.audit_logger = AuditLogger()
    
    async def handle_secure_request(self, request: Request) -> Response:
        # 1. Rate limiting
        client_ip = request.remote
        if not await self.rate_limiter.check_limit(client_ip):
            return self._rate_limit_response()
        
        # 2. Authentication
        auth_header = request.headers.get("Authorization")
        if not await self.auth_validator.validate_token(auth_header):
            return self._unauthorized_response()
        
        # 3. Input validation
        payload = await request.json()
        if not self._validate_payload(payload):
            return self._bad_request_response()
        
        # 4. Audit logging
        await self.audit_logger.log_request(
            client_ip=client_ip,
            method=payload.get("method"),
            timestamp=datetime.utcnow()
        )
        
        # 5. Process request
        return await self._process_request(payload)

# Performance Optimized A2A Client
class OptimizedA2AClient:
    def __init__(self):
        # Connection pooling
        self.session = aiohttp.ClientSession(
            connector=aiohttp.TCPConnector(
                limit=100,
                limit_per_host=20,
                ttl_dns_cache=300,
                enable_cleanup_closed=True
            )
        )
        
        # Response caching
        self.cache = LRUCache(maxsize=1000)
        
        # Request batching
        self.batch_queue = asyncio.Queue()
        self.batch_processor = asyncio.create_task(
            self._process_batches()
        )
    
    async def batch_requests(self, requests: List[Dict]) -> List[Dict]:
        """Batch multiple requests for efficiency"""
        batch_id = str(uuid.uuid4())
        await self.batch_queue.put({
            "batch_id": batch_id,
            "requests": requests
        })
        return await self._wait_for_batch_result(batch_id)

# Resilient A2A Implementation
class ResilientA2AClient:
    def __init__(self):
        self.circuit_breaker = CircuitBreaker(
            failure_threshold=5,
            recovery_timeout=60
        )
        self.retry_policy = ExponentialBackoffRetry(
            max_retries=3,
            base_delay=1.0
        )
        self.fallback_strategy = FallbackStrategy()
    
    @circuit_breaker
    @retry_policy
    async def make_request(self, method: str, params: Dict) -> Dict:
        try:
            return await self._make_actual_request(method, params)
        except Exception as e:
            # Log error
            logger.error(f"Request failed: {e}")
            
            # Try fallback
            return await self.fallback_strategy.execute_fallback(
                method, params
            )

# A2A Monitoring Implementation
class A2AMonitoring:
    def __init__(self):
        self.metrics = {
            "requests_total": Counter(),
            "request_duration": Histogram(),
            "errors_total": Counter(),
            "active_connections": Gauge()
        }
        
        self.tracing = OpenTelemetryTracing()
    
    @contextmanager
    def track_request(self, method: str):
        start_time = time.time()
        self.metrics["requests_total"].inc()
        self.metrics["active_connections"].inc()
        
        with self.tracing.start_span(f"a2a.{method}"):
            try:
                yield
                duration = time.time() - start_time
                self.metrics["request_duration"].observe(duration)
            except Exception as e:
                self.metrics["errors_total"].inc()
                raise
            finally:
                self.metrics["active_connections"].dec()

# docker-compose.yml for A2A deployment
version: '3.8'
services:
  a2a-server:
    build: .
    ports:
      - "8080:8080"
    environment:
      - REDIS_URL=redis://redis:6379
      - LOG_LEVEL=INFO
      - MAX_CONCURRENT_TASKS=100
    depends_on:
      - redis
      - postgres
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8080/health"]
      interval: 30s
      timeout: 10s
      retries: 3
  
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis-data:/data
  
  postgres:
    image: postgres:15-alpine
    environment:
      - POSTGRES_DB=a2a
      - POSTGRES_USER=a2a
      - POSTGRES_PASSWORD=secret
    volumes:
      - postgres-data:/var/lib/postgresql/data

volumes:
  redis-data:
  postgres-data:

# Persistent task storage
class PersistentTaskStorage:
    async def save_task(self, task: Task):
        # Save to both memory and persistent storage
        self.memory_cache[task.id] = task
        await self.redis.set(
            f"task:{task.id}",
            task.json()
        )

# JWT-based authentication
class JWTAuthentication:
    def __init__(self, secret_key: str):
        self.secret_key = secret_key
    
    async def validate_token(self, token: str) -> bool:
        try:
            payload = jwt.decode(
                token,
                self.secret_key,
                algorithms=["HS256"]
            )
            return payload.get("role") in ["agent", "user"]
        except jwt.InvalidTokenError:
            return False

# Token bucket rate limiter
class RateLimiter:
    def __init__(self, tokens_per_minute: int):
        self.tokens = tokens_per_minute
        self.last_refill = time.time()
    
    async def check_limit(self, agent_id: str) -> bool:
        self._refill_tokens()
        if self.tokens > 0:
            self.tokens -= 1
            return True
        return False

# Circuit breaker pattern
class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5):
        self.failures = 0
        self.threshold = failure_threshold
        self.state = "CLOSED"
    
    async def execute(self, func, *args):
        if self.state == "OPEN":
            raise CircuitOpenError()
        
        try:
            result = await func(*args)
            self.failures = 0
            return result
        except Exception:
            self.failures += 1
            if self.failures >= self.threshold:
                self.state = "OPEN"
                asyncio.create_task(self._reset_after_timeout())
            raise