Message Schemas Reference¶

import uuid
import re

def validate_message_id(message_id: str) -> bool:
    """Validate message_id is a valid UUID v4"""
    try:
        uuid_obj = uuid.UUID(message_id, version=4)
        return str(uuid_obj) == message_id
    except (ValueError, AttributeError):
        return False

# Alternative regex validation
UUID_V4_PATTERN = re.compile(
    r'^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$',
    re.IGNORECASE
)

def validate_message_id_regex(message_id: str) -> bool:
    return bool(UUID_V4_PATTERN.match(message_id))

from enum import Enum

class MessageType(Enum):
    """A2A Protocol Message Types"""
    # Discovery
    DISCOVER_AGENTS = "discover_agents"
    AGENT_ANNOUNCEMENT = "agent_announcement"

    # Capability Exchange
    GET_CAPABILITIES = "get_capabilities"
    CAPABILITIES_RESPONSE = "capabilities_response"

    # Request/Response
    REQUEST = "request"
    RESPONSE = "response"
    ERROR = "error"

    # Session Management
    HANDSHAKE = "handshake"
    HANDSHAKE_ACK = "handshake_ack"
    GOODBYE = "goodbye"

    # Streaming
    STREAM_START = "stream_start"
    STREAM_DATA = "stream_data"
    STREAM_END = "stream_end"

def validate_message_type(msg_type: str) -> bool:
    """Validate message_type is a known type"""
    try:
        MessageType(msg_type)
        return True
    except ValueError:
        return False

import re

AGENT_ID_PATTERN = re.compile(r'^[a-zA-Z0-9][a-zA-Z0-9-]{0,126}[a-zA-Z0-9]$')

def validate_agent_id(agent_id: str) -> bool:
    """Validate agent_id format"""
    if not isinstance(agent_id, str):
        return False
    if len(agent_id) < 1 or len(agent_id) > 128:
        return False
    if not AGENT_ID_PATTERN.match(agent_id):
        return False
    return True

def validate_recipient_id(recipient_id: str) -> bool:
    """Validate recipient_id (allows special values)"""
    # Special broadcast/registry values
    if recipient_id in ["*", "registry"]:
        return True
    # Otherwise same rules as agent_id
    return validate_agent_id(recipient_id)

from datetime import datetime, timezone
import re

ISO_8601_PATTERN = re.compile(
    r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d{3})?Z$'
)

def validate_timestamp(timestamp: str) -> bool:
    """Validate ISO 8601 timestamp"""
    if not ISO_8601_PATTERN.match(timestamp):
        return False

    try:
        dt = datetime.fromisoformat(timestamp.replace('Z', '+00:00'))
        return True
    except ValueError:
        return False

def validate_timestamp_freshness(timestamp: str, max_age_seconds: int = 300) -> bool:
    """Validate timestamp is recent (within max_age)"""
    try:
        msg_time = datetime.fromisoformat(timestamp.replace('Z', '+00:00'))
        now = datetime.now(timezone.utc)
        age = (now - msg_time).total_seconds()

        # Reject messages from the future (allow 60s clock skew)
        if age < -60:
            return False

        # Reject messages too old
        if age > max_age_seconds:
            return False

        return True
    except ValueError:
        return False

import json

MAX_PAYLOAD_SIZE = 10 * 1024 * 1024  # 10 MB

def validate_payload_size(payload: dict) -> bool:
    """Validate payload doesn't exceed size limit"""
    try:
        payload_json = json.dumps(payload)
        payload_bytes = payload_json.encode('utf-8')
        return len(payload_bytes) <= MAX_PAYLOAD_SIZE
    except (TypeError, ValueError):
        return False

def validate_request_payload(payload: dict) -> bool:
    """Validate REQUEST message payload"""
    # Required field: method
    if "method" not in payload:
        return False

    if not isinstance(payload["method"], str):
        return False

    if len(payload["method"]) < 1 or len(payload["method"]) > 128:
        return False

    # Optional field: parameters
    if "parameters" in payload:
        if not isinstance(payload["parameters"], dict):
            return False

    return True

def validate_response_payload(payload: dict) -> bool:
    """Validate RESPONSE message payload"""
    # Required field: status
    if "status" not in payload:
        return False

    if payload["status"] not in ["success", "error"]:
        return False

    # If success, must have data
    if payload["status"] == "success":
        if "data" not in payload:
            return False
        if not isinstance(payload["data"], dict):
            return False

    # If error, must have error object
    if payload["status"] == "error":
        if "error" not in payload:
            return False
        if not isinstance(payload["error"], dict):
            return False

        # Error must have code and message
        error = payload["error"]
        if "code" not in error or "message" not in error:
            return False

    return True

import re

SEMVER_PATTERN = re.compile(r'^\d+\.\d+\.\d+(-[a-zA-Z0-9.]+)?$')

def validate_agent_card(agent_card: dict) -> bool:
    """Validate agent card structure"""
    required_fields = [
        "agent_id", "name", "version", "description",
        "capabilities", "supported_protocols"
    ]

    for field in required_fields:
        if field not in agent_card:
            return False

    # Validate agent_id
    if not validate_agent_id(agent_card["agent_id"]):
        return False

    # Validate version (semantic versioning)
    if not SEMVER_PATTERN.match(agent_card["version"]):
        return False

    # Validate capabilities array
    if not isinstance(agent_card["capabilities"], list):
        return False

    for cap in agent_card["capabilities"]:
        if not isinstance(cap, str):
            return False

    # Validate supported_protocols array
    if not isinstance(agent_card["supported_protocols"], list):
        return False

    for protocol in agent_card["supported_protocols"]:
        if not isinstance(protocol, str):
            return False

    return True

def validate_handshake_payload(payload: dict) -> bool:
    """Validate HANDSHAKE message payload"""
    if "agent_card" not in payload:
        return False

    return validate_agent_card(payload["agent_card"])

ERROR_CODE_PATTERN = re.compile(r'^[A-Z][A-Z0-9_]*[A-Z0-9]$')

def validate_error_payload(payload: dict) -> bool:
    """Validate ERROR message payload"""
    if "error" not in payload:
        return False

    error = payload["error"]

    # Required fields
    if "code" not in error or "message" not in error:
        return False

    # Validate code format (UPPERCASE_WITH_UNDERSCORES)
    if not ERROR_CODE_PATTERN.match(error["code"]):
        return False

    # Validate message is non-empty string
    if not isinstance(error["message"], str) or len(error["message"]) < 1:
        return False

    # Validate retry_after if present
    if "retry_after" in error:
        if not isinstance(error["retry_after"], int):
            return False
        if error["retry_after"] < 0:
            return False

    return True

def validate_discover_agents_payload(payload: dict) -> bool:
    """Validate DISCOVER_AGENTS message payload"""
    # All fields are optional

    # Validate status if present
    if "status" in payload:
        if payload["status"] not in ["healthy", "any"]:
            return False

    # Validate limit if present
    if "limit" in payload:
        if not isinstance(payload["limit"], int):
            return False
        if payload["limit"] < 1 or payload["limit"] > 1000:
            return False

    # Validate capability if present
    if "capability" in payload:
        if not isinstance(payload["capability"], str):
            return False

    return True

from typing import Dict, Any, List
from dataclasses import dataclass

@dataclass
class ValidationResult:
    """Result of message validation"""
    valid: bool
    errors: List[str]
    warnings: List[str]

class MessageValidator:
    """Complete A2A message validator"""

    def validate_message(self, message: Dict[str, Any]) -> ValidationResult:
        """
        Validate complete A2A message

        Performs 8-layer validation:
        1. Structure validation (required fields)
        2. Type validation (correct types)
        3. Format validation (UUIDs, timestamps, etc.)
        4. Size validation (payload limits)
        5. Content validation (payload structure)
        6. Business logic validation
        7. Security validation (auth tags if present)
        8. Freshness validation (timestamp age)
        """
        errors = []
        warnings = []

        # Layer 1: Structure validation
        required_fields = [
            "message_id", "message_type", "sender_id",
            "recipient_id", "timestamp", "payload"
        ]

        for field in required_fields:
            if field not in message:
                errors.append(f"Missing required field: {field}")

        if errors:
            return ValidationResult(False, errors, warnings)

        # Layer 2: Type validation
        if not isinstance(message["message_id"], str):
            errors.append("message_id must be string")

        if not isinstance(message["payload"], dict):
            errors.append("payload must be object")

        # Layer 3: Format validation
        if not validate_message_id(message["message_id"]):
            errors.append("Invalid message_id format (not UUID v4)")

        if not validate_message_type(message["message_type"]):
            errors.append(f"Unknown message_type: {message['message_type']}")

        if not validate_agent_id(message["sender_id"]):
            errors.append("Invalid sender_id format")

        if not validate_timestamp(message["timestamp"]):
            errors.append("Invalid timestamp format")

        # Layer 4: Size validation
        if not validate_payload_size(message["payload"]):
            errors.append("Payload exceeds maximum size")

        # Layer 5: Content validation (message type specific)
        msg_type = message["message_type"]
        payload = message["payload"]

        if msg_type == "request":
            if not validate_request_payload(payload):
                errors.append("Invalid request payload")
        elif msg_type == "response":
            if not validate_response_payload(payload):
                errors.append("Invalid response payload")
        elif msg_type == "handshake":
            if not validate_handshake_payload(payload):
                errors.append("Invalid handshake payload")
        # ... other message types

        # Layer 6: Business logic validation
        # (application-specific, extend as needed)

        # Layer 7: Security validation
        if "auth" in message:
            # Validate authentication tag structure
            # (See authentication documentation)
            pass

        # Layer 8: Freshness validation
        if not validate_timestamp_freshness(message["timestamp"]):
            warnings.append("Message timestamp is stale")

        is_valid = len(errors) == 0
        return ValidationResult(is_valid, errors, warnings)

import unittest
import json

class TestMessageValidation(unittest.TestCase):
    """Unit tests for message validation"""

    def test_valid_request_message(self):
        """Test validation of valid REQUEST message"""
        message = {
            "message_id": "a7f8d9e2-3c4b-5d6e-7f8a-9b0c1d2e3f4g",
            "message_type": "request",
            "sender_id": "client-001",
            "recipient_id": "server-001",
            "timestamp": "2025-01-15T10:30:00.000Z",
            "payload": {
                "method": "get_price",
                "parameters": {"currency": "BTC"}
            },
            "correlation_id": None
        }

        validator = MessageValidator()
        result = validator.validate_message(message)

        self.assertTrue(result.valid)
        self.assertEqual(len(result.errors), 0)

    def test_missing_required_field(self):
        """Test validation fails with missing required field"""
        message = {
            "message_id": "a7f8d9e2-3c4b-5d6e-7f8a-9b0c1d2e3f4g",
            "message_type": "request",
            # Missing sender_id
            "recipient_id": "server-001",
            "timestamp": "2025-01-15T10:30:00.000Z",
            "payload": {}
        }

        validator = MessageValidator()
        result = validator.validate_message(message)

        self.assertFalse(result.valid)
        self.assertIn("Missing required field: sender_id", result.errors)

    def test_invalid_uuid_format(self):
        """Test validation fails with invalid UUID"""
        message = {
            "message_id": "not-a-valid-uuid",
            "message_type": "request",
            "sender_id": "client-001",
            "recipient_id": "server-001",
            "timestamp": "2025-01-15T10:30:00.000Z",
            "payload": {}
        }

        validator = MessageValidator()
        result = validator.validate_message(message)

        self.assertFalse(result.valid)
        self.assertIn("Invalid message_id format", result.errors)

# ❌ BAD: Trust user input
def handle_message(message: dict):
    # Process without validation
    method = message["payload"]["method"]
    process(method)

# ✅ GOOD: Validate first
def handle_message(message: dict):
    validator = MessageValidator()
    result = validator.validate_message(message)

    if not result.valid:
        return create_error_response(
            "INVALID_MESSAGE",
            f"Validation failed: {'; '.join(result.errors)}"
        )

    # Now safe to process
    method = message["payload"]["method"]
    process(method)

# ✅ Use helper functions
def create_request_message(
    sender_id: str,
    recipient_id: str,
    method: str,
    parameters: dict = None
) -> dict:
    """Create a valid REQUEST message"""
    return {
        "message_id": str(uuid.uuid4()),
        "message_type": "request",
        "sender_id": sender_id,
        "recipient_id": recipient_id,
        "timestamp": datetime.now(timezone.utc).isoformat().replace('+00:00', 'Z'),
        "payload": {
            "method": method,
            "parameters": parameters or {}
        },
        "correlation_id": None
    }

import jsonschema

# Define JSON Schema
REQUEST_SCHEMA = {
    "type": "object",
    "required": ["message_id", "message_type", "sender_id", "recipient_id", "timestamp", "payload"],
    "properties": {
        "message_id": {"type": "string", "pattern": "^[0-9a-f-]{36}$"},
        "message_type": {"type": "string", "enum": ["request"]},
        "sender_id": {"type": "string", "minLength": 1, "maxLength": 128},
        "recipient_id": {"type": "string", "minLength": 1, "maxLength": 128},
        "timestamp": {"type": "string", "format": "date-time"},
        "payload": {
            "type": "object",
            "required": ["method"],
            "properties": {
                "method": {"type": "string", "minLength": 1},
                "parameters": {"type": "object"}
            }
        },
        "correlation_id": {"type": ["string", "null"]}
    }
}

def validate_with_json_schema(message: dict) -> bool:
    """Validate message using JSON Schema"""
    try:
        jsonschema.validate(instance=message, schema=REQUEST_SCHEMA)
        return True
    except jsonschema.ValidationError as e:
        print(f"Validation error: {e.message}")
        return False