Skip to content

Agent Registry A2A Protocol Integration

Date: 2025-04-26 Participants: AI Assistant, Rakesh Gangwar

Context

Building on our initial Agent Registry implementation, we've enhanced the agent model to support the A2A (Agent-to-Agent) protocol and improved the overall agent schema to better handle different agent types and capabilities.

Discussion

A2A Protocol Support

  1. A2A Protocol Requirements:
  2. Standard protocol for interoperability between agents from different systems
  3. Requires specific metadata and capabilities description
  4. Enables discovery and communication between agents

  5. Supports various interaction modes (text, files, data)

  6. A2A Agent Card Format:

  7. Name, description, version (already supported)
  8. URL for agent endpoint (new)
  9. Provider information (new)
  10. Capabilities object with streaming, push notifications, history support (new)
  11. Authentication details (new)
  12. Input/output modes (similar to our modalities)

  13. Skills array with detailed abilities (new)

  14. Schema Comparison:

  15. Identified differences between our model and A2A requirements
  16. Found inconsistencies between backend model and frontend mock data
  17. Consolidated all requirements into a unified schema

Implementation Details

  1. Enhanced Agent Model:
  2. Added type classification with execution_type field
  3. Added status field (active, inactive, deprecated)
  4. Added last_used timestamp for usage tracking
  5. Added capabilities array to complement modalities

  6. Added A2A protocol support fields:

    • URL, provider details, authentication details
    • Capability flags for streaming, push, history

  7. Agent Skills Model:

  8. Created new table for structured skill representation
  9. Supports detailed descriptions of agent capabilities
  10. Includes examples and I/O modes per skill

  11. Maps to A2A protocol skills concept

  12. Schema Updates: 

    # New fields for Agent model
execution_type = Column(String(100))  # docker, api, a2a, framework, edge
status = Column(String(50), default="active")
last_used = Column(DateTime)
capabilities = Column(JSON)
url = Column(String(255))
provider_name = Column(String(100))
provider_url = Column(String(255))
authentication_type = Column(String(50))
authentication_config = Column(JSON)
capabilities_streaming = Column(Boolean, default=False)
capabilities_push = Column(Boolean, default=False)
capabilities_history = Column(Boolean, default=False)

  13. Agent Skills Schema: 

    class AgentSkill(Base):
    id = Column(UUID, primary_key=True)
    agent_id = Column(UUID, ForeignKey("agents.id"))
    name = Column(String(100), nullable=False)
    description = Column(Text)
    tags = Column(JSON)
    examples = Column(JSON)
    input_modes = Column(JSON)
    output_modes = Column(JSON)

Challenges & Solutions

  1. Schema Differences:
  2. Found inconsistency between backend and frontend models
  3. Type field had different values (e.g., "text" vs. "llm")

  4. Solution: Added type mapping for frontend display

  5. A2A Integration:

  6. Added well-known endpoint pattern for A2A Agent Card
  7. Created format conversion between our schema and A2A format

  8. Maintained backward compatibility with existing code

  9. Field Normalization:

  10. Organized fields into logical groups
  11. Ensured consistent naming conventions
  12. Added proper documentation and comments

Implementation Progress

A2A Protocol Endpoints

  1. Well-Known Endpoint:
  2. Implemented GET /agents/{agent_id}/.well-known/agent.json endpoint
  3. Dynamically generates A2A Agent Card from our internal model
  4. Handles conversion between our data model and A2A protocol format

  5. Exposes agent skills and capabilities in the standard format

  6. A2A Task Management Endpoints:

  7. Implemented basic A2A protocol endpoints for task interaction:

    • POST /a2a/{agent_id}/tasks/send - Send message to initiate/continue a task
    • POST /a2a/{agent_id}/tasks/sendSubscribe - Send message with SSE streaming
    • POST /a2a/{agent_id}/tasks/get - Get current task state
    • POST /a2a/{agent_id}/tasks/cancel - Cancel a running task
    • POST /a2a/{agent_id}/tasks/pushNotification/set - Configure push notifications
    • POST /a2a/{agent_id}/tasks/pushNotification/get - Get push notification config

  8. Streaming Support:

  9. Added Server-Sent Events (SSE) implementation for real-time updates
  10. Supports incremental responses from agents

  11. Provides structured message format for streaming interactions

  12. Authentication Integration:

  13. Added get_current_user_id function to extract UUID from user token
  14. Ensured authenticated access to A2A endpoints
  15. Maintains security while enabling agent interoperability

Technical Details

  1. Message Format: 

    class Message(BaseModel):
    role: str = Field(..., description="Message role (user, agent)")
    parts: list[MessagePart] = Field(..., description="Message parts")
    metadata: Optional[Dict[str, Any]] = Field(None, description="Message metadata")

class MessagePart(BaseModel):
    type: str = Field(..., description="Part type (text, file, data)")
    text: Optional[str] = Field(None, description="Text content")
    file: Optional[Dict[str, Any]] = Field(None, description="File content")
    data: Optional[Dict[str, Any]] = Field(None, description="Structured data")

  2. SSE Implementation: 

    @router.post("/a2a/{agent_id}/tasks/sendSubscribe")
async def send_task_subscribe(agent_id: UUID, params: TaskSendParams):
    # This is a simplified StreamingResponse example
    async def event_generator():
        yield format_sse_event(json.dumps({
            "id": "task-123",
            "status": { "state": "working", "timestamp": "..." }
        }))
        # Additional events...

    return StreamingResponse(event_generator(), media_type="text/event-stream")

  3. A2A Card Generation: 

    agent_card = {
    "name": agent.name,
    "description": agent.description,
    "url": agent.url or agent_url,
    "provider": { "name": agent.provider_name, "url": agent.provider_url },
    "capabilities": {
        "streaming": agent.capabilities_streaming,
        "pushNotifications": agent.capabilities_push,
        "stateTransitionHistory": agent.capabilities_history
    },
    "skills": [ /* skill mappings */ ]
}

Decision

  1. Unified Agent Schema:
  2. Consolidated backend model, frontend expectations, and A2A requirements
  3. Created complete schema documentation

  4. Added clear comments for all fields

  5. Frontend Integration:

  6. Planned API service for agent operations
  7. Created type mapping helpers for frontend display

  8. Ensured consistent data format across all components

  9. A2A Protocol Support:

  10. Implemented fields needed for A2A protocol compatibility
  11. Added support for skills representation
  12. Created A2A Card endpoint format
  13. Implemented core A2A protocol endpoints

Next Steps

  1. Remaining API Implementation:
  2. Add endpoints for agent skills CRUD operations
  3. Enable direct skill management through API

  4. Implement full task persistence and management

  5. Frontend Updates:

  6. Create agentApi.ts service
  7. Update agent-store.ts to use backend API

  8. Add UI for editing agent skills and A2A fields

  9. Testing:

  10. Add tests for new schema validation
  11. Validate A2A Card generation
  12. Test type mapping and conversions

  13. Verify authenticated endpoint access

  14. Documentation:

  15. Update API documentation with new fields
  16. Document A2A protocol integration
  17. Create guide for agent skills management
  18. Add usage examples for A2A protocol endpoints

References