> ## Documentation Index
> Fetch the complete documentation index at: https://docs.hexr.dev/llms.txt
> Use this file to discover all available pages before exploring further.
# hexr.a2a
> Agent-to-Agent communication with JSON-RPC 2.0 protocol. Discover remote agents, send messages, stream responses, and manage task lifecycle — all over mTLS.
## Quick Start
### Sending a Message
```python theme={null}
from hexr.a2a import A2AClient, Message, TextPart
async with A2AClient("http://content-crew-a2a.tenant-acme.svc:8090") as client:
# Discover the remote agent
card = await client.discover()
print(f"Agent: {card.name}, Skills: {[s.name for s in card.skills]}")
# Send a message
task = await client.send(Message(
parts=[TextPart(text="Write a blog post about AI agents")]
))
print(f"Task: {task.id}, State: {task.state}")
# Task: task_abc123, State: completed
for artifact in task.artifacts:
print(artifact.parts[0].text)
```
### Receiving Messages (A2A Bridge)
When you set `a2a=True` on `@hexr_agent`, the bridge is automatic:
```python theme={null}
@hexr_agent(
name="content-crew",
tenant="acme-corp",
a2a=True,
skills=[{"id": "write", "name": "Write", "description": "Write content"}]
)
def handle_request(message: str) -> str:
# This function is called when another agent sends a message
return f"Here's your content about: {message}"
```
***
## A2AClient
```python theme={null}
from hexr.a2a import A2AClient
client = A2AClient(
base_url="http://agent-name-a2a.tenant-namespace.svc:8090",
timeout=300, # Max wait for task completion
max_retries=3, # Retry on transient failures
card_cache_ttl=300 # Cache agent card for 5 minutes
)
```
### Methods
Fetch the remote agent's Agent Card.
```python theme={null}
card = await client.discover()
print(card.name) # "content-crew"
print(card.description) # "Multi-agent content creation crew"
print(card.skills) # [Skill(id="write", name="Write", ...)]
print(card.capabilities) # {streaming: true, ...}
```
Returns: `AgentCard`
Send a message to the remote agent. Blocks until the task reaches a terminal state.
```python theme={null}
task = await client.send(Message(
parts=[TextPart(text="Analyze Q4 financials")]
))
if task.state == TaskState.COMPLETED:
result = task.artifacts[0].parts[0].text
elif task.state == TaskState.FAILED:
error = task.artifacts[0].parts[0].text
```
Returns: `Task`
Poll a task's current state.
```python theme={null}
task = await client.get_task("task_abc123")
print(task.state) # submitted | working | completed | failed | canceled
```
Request cooperative cancellation of a task.
```python theme={null}
task = await client.cancel_task("task_abc123")
# The agent should check for cancellation and stop gracefully
```
***
## Data Models
### Message
```python theme={null}
from hexr.a2a import Message, TextPart, DataPart, FilePart, FileContent
# Text message
msg = Message(parts=[TextPart(text="Hello")])
# Structured data
msg = Message(parts=[DataPart(data={"query": "analyze", "sector": "tech"})])
# File attachment
msg = Message(parts=[FilePart(file=FileContent(
name="report.pdf",
mimeType="application/pdf",
bytes="base64-encoded-content"
))])
```
### Task States
```python theme={null}
from hexr.a2a import TaskState
# Lifecycle: submitted → working → completed | failed | canceled
TaskState.SUBMITTED # Task received, not yet started
TaskState.WORKING # Agent is processing
TaskState.COMPLETED # Successfully finished (has artifacts)
TaskState.FAILED # Error occurred
TaskState.CANCELED # Cooperative cancellation
TaskState.INPUT_REQUIRED # Agent needs more info (has question in artifacts)
```
### AgentCard
```python theme={null}
card = await client.discover()
card.name # Agent name
card.description # Human-readable description
card.url # Agent A2A endpoint URL
card.version # Agent version
card.skills # List of skills [{id, name, description}]
card.capabilities # {streaming, pushNotifications, stateTransitionHistory}
```
***
## Patterns
### Fan-Out / Fan-In (Orchestrator)
```python theme={null}
@hexr_agent(name="orchestrator", tenant="acme-corp", a2a=True)
async def orchestrate(question: str) -> str:
import asyncio
# Fan-out: call multiple agents in parallel
async with A2AClient("http://researcher-a2a.tenant-acme.svc:8090") as researcher, \
A2AClient("http://analyst-a2a.tenant-acme.svc:8090") as analyst:
research_task, analysis_task = await asyncio.gather(
researcher.send(Message(parts=[TextPart(text=question)])),
analyst.send(Message(parts=[TextPart(text=question)]))
)
# Fan-in: combine results
research = research_task.artifacts[0].parts[0].text
analysis = analysis_task.artifacts[0].parts[0].text
client = hexr_llm(openai.OpenAI())
synthesis = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": f"Synthesize:\n{research}\n{analysis}"}]
)
return synthesis.choices[0].message.content
```
### Pipeline (Sequential)
```python theme={null}
@hexr_agent(name="pipeline", tenant="acme-corp", a2a=True)
async def pipeline(data: str) -> str:
# Step 1: Process
async with A2AClient("http://processor-a2a.svc:8090") as proc:
processed = await proc.send(Message(parts=[TextPart(text=data)]))
# Step 2: Review
async with A2AClient("http://reviewer-a2a.svc:8090") as rev:
reviewed = await rev.send(Message(
parts=[TextPart(text=processed.artifacts[0].parts[0].text)]
))
return reviewed.artifacts[0].parts[0].text
```
***
## Discovery
Agents are discovered via DNS + Agent Cards:
```
# DNS: {agent-name}-a2a.{namespace}.svc.cluster.local
http://content-crew-a2a.tenant-acme-corp.svc:8090
# Agent Card: served by Envoy from ConfigMap
GET /.well-known/agent.json → AgentCard JSON
```
All A2A communication is over **mTLS** via Envoy sidecars. Both parties must have valid SPIFFE identities.
***
## Task State Storage
Tasks are stored in **Valkey** (Redis-compatible) with:
* **SETNX** for idempotent task creation
* **TTL** for automatic cleanup (configurable, default 24h)
* **Task Reaper** background goroutine for stuck tasks
* **Cooperative cancellation** via cancel flags