A2A Protocol for AI Agents — How We Lost $40k to Agent Handshake Timeouts

import requests
import json
from typing import Dict, Any

# A2A handshake implementation
# We use requests.Session to reuse connections and avoid re-handshake
session = requests.Session()
session.headers.update({'Content-Type': 'application/json'})

def perform_handshake(target_url: str, capabilities: list) -> str:
    """Returns a session token if handshake succeeds."""
    # Step 1: Send our capabilities and request target's capabilities
    handshake_payload = {
        "agent_id": "primary-agent-v2",
        "capabilities": capabilities,
        "requested_capabilities": ["profile", "demographics"]  # what we need from target
    }
    try:
        # Timeout is critical: we set it to 15s after the incident
        resp = session.post(f"{target_url}/a2a/handshake", json=handshake_payload, timeout=15)
        resp.raise_for_status()
        data = resp.json()
        # Step 2: Validate that target supports what we need
        if not set(["profile", "demographics"]).issubset(data.get("supported_capabilities", [])):
            raise ValueError("Target agent missing required capabilities")
        return data["session_token"]  # used for subsequent task calls
    except requests.exceptions.Timeout:
        # This is what we saw in logs: A2AHandshakeTimeout
        raise RuntimeError("A2A handshake timed out after 15s")

# Usage
if __name__ == "__main__":
    token = perform_handshake("http://user-profile-agent:8080", ["recommendation"])
    print(f"Handshake succeeded, token: {token[:8]}...")

from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel
import uuid
import time
from typing import Dict, Any

# A2A protocol library (install: pip install a2a-protocol==0.2.1)
from a2a_protocol import A2AHandler, HandshakeRequest, TaskRequest

app = FastAPI()
handler = A2AHandler()

# In-memory session store (use Redis in production)
sessions: Dict[str, float] = {}  # token -> creation timestamp
MAX_SESSION_AGE = 300  # 5 minutes

class AgentCapabilities(BaseModel):
    agent_id: str
    capabilities: list[str]
    requested_capabilities: list[str]

@app.post("/a2a/handshake")
async def handshake(req: AgentCapabilities):
    # Validate caller capabilities
    if "recommendation" not in req.capabilities:
        raise HTTPException(status_code=403, detail="Caller must have 'recommendation' capability")
    # Generate session token
    token = str(uuid.uuid4())
    sessions[token] = time.time()
    return {
        "session_token": token,
        "supported_capabilities": ["profile", "demographics"],
        "session_ttl_seconds": MAX_SESSION_AGE
    }

class TaskPayload(BaseModel):
    session_token: str
    task_type: str
    data: Dict[str, Any]

@app.post("/a2a/task")
async def task(req: TaskPayload):
    # Validate session token
    if req.session_token not in sessions:
        raise HTTPException(status_code=401, detail="Invalid or expired session token")
    # Check session age
    if time.time() - sessions[req.session_token] > MAX_SESSION_AGE:
        del sessions[req.session_token]
        raise HTTPException(status_code=401, detail="Session token expired")
    # Process task
    if req.task_type == "profile":
        # Simulate profile enrichment
        return {"status": "success", "profile": {"user_id": req.data["user_id"], "name": "John Doe"}}
    else:
        raise HTTPException(status_code=400, detail=f"Unknown task type: {req.task_type}")

# Health check endpoint for monitoring
@app.get("/health")
async def health():
    return {"status": "ok", "active_sessions": len(sessions)}

import asyncio
import random
from typing import Dict, List

# Exponential backoff for handshake retries
async def handshake_with_backoff(target_url: str, capabilities: list, max_retries: int = 5):
    base_delay = 0.1  # 100ms
    for attempt in range(max_retries):
        try:
            # Perform handshake (omitted for brevity)
            return await perform_handshake(target_url, capabilities)
        except RuntimeError:
            delay = base_delay * (2 ** attempt) + random.uniform(0, 0.1)
            print(f"Handshake attempt {attempt+1} failed, retrying in {delay:.2f}s")
            await asyncio.sleep(delay)
    raise RuntimeError("Handshake failed after max retries")

# Consistent hash ring for task routing
class ConsistentHashRing:
    def __init__(self, nodes: List[str], replicas: int = 3):
        self.replicas = replicas
        self.ring: Dict[int, str] = {}
        for node in nodes:
            for i in range(replicas):
                key = hash(f"{node}:{i}")
                self.ring[key] = node
    
    def get_node(self, task_key: str) -> str:
        if not self.ring:
            raise ValueError("No nodes in ring")
        key = hash(task_key)
        sorted_keys = sorted(self.ring.keys())
        for k in sorted_keys:
            if key <= k:
                return self.ring[k]
        return self.ring[sorted_keys[0]]  # wrap around

# Usage
ring = ConsistentHashRing(["agent-1", "agent-2", "agent-3"])
task_type = "profile"
target_agent = ring.get_node(task_type)
print(f"Routing {task_type} to {target_agent}")

import asyncio
import httpx
from fastapi import FastAPI

app = FastAPI()

# Mistake #3: Blocking I/O in handshake (WRONG)
@app.post("/a2a/handshake_wrong")
async def handshake_wrong():
    import requests
    # This blocks the event loop!
    resp = requests.get("http://external-api/capabilities", timeout=5)
    return {"capabilities": resp.json()}

# Correct: async I/O
async def fetch_capabilities():
    async with httpx.AsyncClient() as client:
        resp = await client.get("http://external-api/capabilities", timeout=5)
        return resp.json()

@app.post("/a2a/handshake_correct")
async def handshake_correct():
    capabilities = await fetch_capabilities()
    return {"capabilities": capabilities}

import structlog
from prometheus_client import Histogram, Counter, generate_latest
from fastapi import FastAPI, Response

app = FastAPI()
logger = structlog.get_logger()

# Prometheus metrics
handshake_duration = Histogram('a2a_handshake_duration_seconds', 'Duration of A2A handshake', buckets=[0.1, 0.5, 1.0, 2.0, 5.0])
task_duration = Histogram('a2a_task_duration_seconds', 'Duration of A2A task', buckets=[0.5, 1.0, 2.0, 5.0, 10.0])
errors = Counter('a2a_errors_total', 'Total A2A errors', ['error_code'])

@app.post("/a2a/handshake")
async def handshake():
    with handshake_duration.time():
        # ... handshake logic ...
        logger.info("handshake_completed", agent_id="primary", session_token="abc123")
        return {"session_token": "abc123"}

@app.get("/metrics")
async def metrics():
    return Response(content=generate_latest(), media_type="text/plain")

# Debug endpoint
active_sessions = {}  # In production, use Redis
@app.get("/debug/a2a/sessions")
async def list_sessions():
    return {"active_sessions": len(active_sessions), "sessions": list(active_sessions.keys())}