Error Handling in A2A Protocol¶

async def handle_validation_error(error_msg: dict):
    """Handle validation errors - don't retry, fix the request"""
    error = error_msg["payload"]["error"]

    # Log for debugging
    logger.error(f"Validation failed: {error['message']}")
    logger.debug(f"Details: {error['details']}")

    # Don't retry - validation errors won't fix themselves
    # Application logic should fix the request structure
    raise ValueError(f"Invalid request: {error['message']}")

import asyncio
from datetime import datetime, timedelta

async def handle_rate_limit_error(error_msg: dict):
    """Handle rate limit with exponential backoff"""
    error = error_msg["payload"]["error"]
    retry_after = error.get("retry_after", 60)

    logger.warning(f"Rate limited. Waiting {retry_after}s before retry")

    # Honor retry_after with small buffer
    await asyncio.sleep(retry_after + 1)

    # After waiting, retry the original request
    # Your retry logic here

async def handle_service_unavailable(
    error_msg: dict,
    max_retries: int = 3
):
    """Handle service unavailable with retry logic"""
    error = error_msg["payload"]["error"]
    retry_after = error.get("retry_after", 60)

    for attempt in range(max_retries):
        logger.info(f"Service unavailable. Retry {attempt+1}/{max_retries}")

        # Exponential backoff: retry_after * 2^attempt
        wait_time = retry_after * (2 ** attempt)
        await asyncio.sleep(wait_time)

        # Attempt retry
        try:
            return await retry_request()
        except ServiceUnavailableError:
            if attempt == max_retries - 1:
                raise
            continue

async def handle_authentication_error(error_msg: dict):
    """Handle authentication failures"""
    error = error_msg["payload"]["error"]

    logger.error(f"Authentication failed: {error['message']}")

    # Check if it's a signature issue
    if "signature" in error["message"].lower():
        # Regenerate signature with correct algorithm
        await regenerate_credentials()

    # Check if token expired
    elif "expired" in error["message"].lower():
        # Re-authenticate
        await reauthenticate()

    else:
        # Unknown auth error - don't retry, requires manual fix
        raise AuthenticationError(error["message"])

import asyncio
import random

async def exponential_backoff_retry(
    func,
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    jitter: bool = True
):
    """
    Retry with exponential backoff

    Args:
        func: Async function to retry
        max_retries: Maximum retry attempts
        base_delay: Initial delay in seconds
        max_delay: Maximum delay cap
        jitter: Add randomness to prevent thundering herd
    """
    for attempt in range(max_retries):
        try:
            return await func()
        except RetryableError as e:
            if attempt == max_retries - 1:
                raise

            # Calculate delay: base_delay * 2^attempt
            delay = min(base_delay * (2 ** attempt), max_delay)

            # Add jitter (randomness) to prevent thundering herd
            if jitter:
                delay = delay * (0.5 + random.random())

            logger.info(f"Retry {attempt+1}/{max_retries} after {delay:.2f}s")
            await asyncio.sleep(delay)

# Retry a request with exponential backoff
result = await exponential_backoff_retry(
    lambda: send_a2a_request(agent_id, message),
    max_retries=5,
    base_delay=1.0
)

from enum import Enum
from datetime import datetime, timedelta

class CircuitState(Enum):
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Failures detected, blocking requests
    HALF_OPEN = "half_open"  # Testing if service recovered

class CircuitBreaker:
    """
    Circuit breaker for agent communication

    States:
    - CLOSED: Normal operation, requests pass through
    - OPEN: Too many failures, blocking all requests
    - HALF_OPEN: Testing recovery, allowing limited requests
    """

    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: float = 60.0,
        success_threshold: int = 2
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.success_threshold = success_threshold

        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time = None

    async def call(self, func):
        """Execute function through circuit breaker"""

        # If OPEN, check if recovery timeout passed
        if self.state == CircuitState.OPEN:
            if self._should_attempt_reset():
                self.state = CircuitState.HALF_OPEN
                self.success_count = 0
            else:
                raise CircuitBreakerOpenError(
                    f"Circuit breaker OPEN. Try again after "
                    f"{self.recovery_timeout}s"
                )

        try:
            result = await func()
            self._on_success()
            return result

        except Exception as e:
            self._on_failure()
            raise

    def _on_success(self):
        """Handle successful request"""
        if self.state == CircuitState.HALF_OPEN:
            self.success_count += 1

            # Enough successes? Close circuit
            if self.success_count >= self.success_threshold:
                self.state = CircuitState.CLOSED
                self.failure_count = 0
                logger.info("Circuit breaker CLOSED (recovered)")

        elif self.state == CircuitState.CLOSED:
            # Reset failure count on success
            self.failure_count = 0

    def _on_failure(self):
        """Handle failed request"""
        self.failure_count += 1
        self.last_failure_time = datetime.utcnow()

        if self.state == CircuitState.HALF_OPEN:
            # Failure during recovery - back to OPEN
            self.state = CircuitState.OPEN
            logger.warning("Circuit breaker re-OPENED (recovery failed)")

        elif self.failure_count >= self.failure_threshold:
            # Too many failures - open circuit
            self.state = CircuitState.OPEN
            logger.error("Circuit breaker OPENED (too many failures)")

    def _should_attempt_reset(self) -> bool:
        """Check if enough time passed to attempt reset"""
        if not self.last_failure_time:
            return True

        elapsed = (datetime.utcnow() - self.last_failure_time).total_seconds()
        return elapsed >= self.recovery_timeout

# Usage
breaker = CircuitBreaker(
    failure_threshold=5,
    recovery_timeout=60.0
)

try:
    result = await breaker.call(
        lambda: send_a2a_request(agent_id, message)
    )
except CircuitBreakerOpenError:
    logger.error("Agent unavailable - circuit breaker open")
    # Use fallback or cached data

# DON'T DO THIS
try:
    user = authenticate(username, password)
except Exception as e:
    return {
        "error": str(e),  # Might leak: "User 'admin' not found in database"
        "stack_trace": traceback.format_exc(),  # Exposes code structure
        "database": "postgresql://prod-db-1.internal:5432/users"  # Leaks infrastructure
    }

# DO THIS
try:
    user = authenticate(username, password)
except AuthenticationError as e:
    # Generic message to client
    return {
        "error": {
            "code": "AUTHENTICATION_FAILED",
            "message": "Authentication failed"  # No details about why
        }
    }
    # Detailed logging server-side only
    logger.error(f"Auth failed: {username} - {str(e)}")

class ErrorResponse:
    """Dual error messages: detailed for logs, safe for clients"""

    def __init__(self, code: str, internal_msg: str, client_msg: str):
        self.code = code
        self.internal_message = internal_msg
        self.client_message = client_msg

    def to_client(self) -> dict:
        """Safe message for external consumption"""
        return {
            "error": {
                "code": self.code,
                "message": self.client_message
            }
        }

    def to_log(self) -> str:
        """Detailed message for internal logs"""
        return f"[{self.code}] {self.internal_message}"

# Usage
try:
    process_file(filepath)
except FileNotFoundError as e:
    error = ErrorResponse(
        code="FILE_NOT_FOUND",
        internal_msg=f"File not found: {filepath} (user: {user_id})",
        client_msg="Requested file not found"
    )

    logger.error(error.to_log())  # Detailed logging
    return error.to_client()       # Safe client response

from collections import defaultdict
from datetime import datetime, timedelta

class ErrorRateLimiter:
    """Rate limit error responses to prevent information leakage"""

    def __init__(self, max_errors_per_minute: int = 10):
        self.max_errors = max_errors_per_minute
        self.error_counts = defaultdict(list)  # agent_id -> [timestamps]

    def should_send_error(self, agent_id: str) -> bool:
        """Check if we should send detailed error"""
        now = datetime.utcnow()
        cutoff = now - timedelta(minutes=1)

        # Clean old timestamps
        self.error_counts[agent_id] = [
            ts for ts in self.error_counts[agent_id]
            if ts > cutoff
        ]

        # Check limit
        if len(self.error_counts[agent_id]) >= self.max_errors:
            return False

        self.error_counts[agent_id].append(now)
        return True

# Usage
error_limiter = ErrorRateLimiter(max_errors_per_minute=10)

if not error_limiter.should_send_error(agent_id):
    # Too many errors - send generic response
    return {
        "error": {
            "code": "RATE_LIMIT_EXCEEDED",
            "message": "Too many errors. Please contact support."
        }
    }
else:
    # Send specific error
    return detailed_error_response

import pytest

@pytest.mark.asyncio
async def test_validation_error_handling():
    """Test handling of validation errors"""
    # Send invalid message
    message = {
        "message_id": str(uuid.uuid4()),
        "message_type": "request",
        # Missing required fields...
    }

    response = await send_message(message)

    # Should get validation error
    assert response["message_type"] == "error"
    assert response["payload"]["error"]["code"] == "VALIDATION_FAILED"
    assert "correlation_id" in response

@pytest.mark.asyncio
async def test_rate_limit_retry():
    """Test retry logic for rate limits"""
    # Send many requests to trigger rate limit
    for i in range(150):
        response = await send_request()

    # Should get rate limit error
    assert response["payload"]["error"]["code"] == "RATE_LIMIT_EXCEEDED"
    assert "retry_after" in response["payload"]["error"]

    # Wait and retry should succeed
    await asyncio.sleep(response["payload"]["error"]["retry_after"])
    retry_response = await send_request()
    assert retry_response["message_type"] != "error"

@pytest.mark.asyncio
async def test_circuit_breaker_opens():
    """Test circuit breaker opens after failures"""
    breaker = CircuitBreaker(failure_threshold=3)

    # Simulate failures
    for i in range(3):
        with pytest.raises(Exception):
            await breaker.call(lambda: failing_request())

    # Circuit should be open
    assert breaker.state == CircuitState.OPEN

    # Next call should fail immediately
    with pytest.raises(CircuitBreakerOpenError):
        await breaker.call(lambda: working_request())

@pytest.mark.asyncio
async def test_no_sensitive_data_in_errors():
    """Ensure errors don't leak sensitive data"""
    # Trigger various errors
    errors = [
        await trigger_auth_error(),
        await trigger_validation_error(),
        await trigger_internal_error()
    ]

    # Check none contain sensitive data
    sensitive_patterns = [
        r'\b\d{3}-\d{2}-\d{4}\b',  # SSN
        r'password',
        r'secret',
        r'token',
        r'/var/',  # File paths
        r'postgresql://'  # Connection strings
    ]

    for error in errors:
        error_str = json.dumps(error)
        for pattern in sensitive_patterns:
            assert not re.search(pattern, error_str, re.IGNORECASE)

from prometheus_client import Counter, Histogram

# Error counters by type
error_counter = Counter(
    'a2a_errors_total',
    'Total A2A errors',
    ['agent_id', 'error_code', 'error_type']
)

# Retry attempts
retry_counter = Counter(
    'a2a_retries_total',
    'Total retry attempts',
    ['agent_id', 'attempt_number']
)

# Circuit breaker state
circuit_breaker_state = Gauge(
    'a2a_circuit_breaker_state',
    'Circuit breaker state (0=closed, 1=open, 2=half_open)',
    ['agent_id']
)

# Error response time
error_response_time = Histogram(
    'a2a_error_response_seconds',
    'Time to generate error response',
    ['error_code']
)

# Usage in code
with error_response_time.labels(error_code='RATE_LIMIT_EXCEEDED').time():
    error_response = generate_error_response(...)
    error_counter.labels(
        agent_id=agent_id,
        error_code='RATE_LIMIT_EXCEEDED',
        error_type='client'
    ).inc()