Skip to main content

transports.a2a

A2A (Agent-to-Agent) protocol adapter.

Implements the A2A protocol for agent-to-agent communication. Supports identity context for OAuth token propagation across agent boundaries.

A2ATransport Objects​

class A2ATransport(BaseTransport)

A2A (Agent-to-Agent) protocol adapter.

Implements the A2A protocol for inter-agent communication. This protocol enables agents to communicate and collaborate.

Protocol Features:

  • Standardized message format for agent communication
  • Task delegation support
  • Result aggregation
  • Capability negotiation

Example:

```python
from agent_runtimes.agents import PydanticAIAgent
from agent_runtimes.transports import A2ATransport

agent = PydanticAIAgent(...)
adapter = A2ATransport(agent)

# Handle a request from another agent
response = await adapter.handle_request({
    "task": "analyze_data",
    "data": {"values": [1, 2, 3]},
    "sender_agent_id": "agent-456",
    "conversation_id": "conv-789"
})
```

protocol_name​

@property
def protocol_name() -> str

Get the protocol name.

handle_request​

async def handle_request(request: dict[str, Any]) -> dict[str, Any]

Handle an A2A request.

Arguments:

  • request - A2A request data with keys:
    • task: Task description
    • data: Task data/parameters
    • sender_agent_id: ID of the requesting agent
    • conversation_id: Conversation identifier
    • capabilities_required: Optional list of required capabilities
    • identities: Optional OAuth identities (list of {provider, accessToken})

Returns:

A2A response data with keys:

  • result: Task result
  • status: Success/failure status
  • sender_agent_id: This agent's ID
  • receiver_agent_id: Original sender's ID
  • conversation_id: Conversation identifier
  • metadata: Additional response data

handle_stream​

async def handle_stream(
        request: dict[str, Any]) -> AsyncIterator[dict[str, Any]]

Handle a streaming A2A request.

Arguments:

  • request - A2A request data with optional identities key for OAuth tokens.

Yields:

A2A stream events with keys:

  • type: Event type (progress, result, error)
  • data: Event data
  • conversation_id: Conversation identifier
  • sender_agent_id: This agent's ID