A2A (Agent-to-Agent) is Google's open protocol for cross-platform agent communication. VIBE AI implements A2A to enable agents to collaborate across different frameworks and platforms.
Every A2A-compatible agent publishes a discovery document:
// /.well-known/agent-card.json
{
"name": "VIBE Research Agent",
"description": "AI research analyst specializing in DeFi",
"version": "1.0.0",
"endpoint": "https://agent.vibe.airforce/a2a",
"capabilities": {
"skills": [
{
"name": "research",
"description": "Research any topic",
"input_schema": {...}
},
{
"name": "analyze",
"description": "Analyze data",
"input_schema": {...}
}
],
"streaming": true,
"push_notifications": true
},
"security": {
"schemes": ["bearer", "x402"]
}
}
┌─────────────────────────────────────────────────────────────────────┐
│ A2A TASK LIFECYCLE │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │submitted │───▶│ working │───▶│completed │ │ failed │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
│ │ │ ▲ │
│ │ │ │ │
│ │ ▼ │ │
│ │ ┌──────────┐ │ │
│ │ │ input │ │ │
│ │ │ required │─────────────────────────┘ │
│ │ └──────────┘ │
│ │ │ │
│ │ ▼ │
│ │ ┌──────────┐ │
│ └────────▶│cancelled │ │
│ └──────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘
{
"message_id": "msg_123",
"task_id": "task_456",
"role": "agent",
"parts": [
{
"type": "text",
"content": "Here is the analysis..."
},
{
"type": "file",
"name": "report.pdf",
"content_type": "application/pdf",
"data": "base64..."
}
],
"metadata": {
"model": "gpt-4o",
"tokens_used": 1500
}
}
# Agent configuration
a2a:
enabled: true
endpoint: "/.well-known/agent-card.json"
skills:
- name: "research"
description: "Research any topic"
handler: "research_skill"
- name: "analyze"
description: "Analyze data"
handler: "analyze_skill"
security:
require_auth: true
schemes: ["bearer", "x402"]
# A2A task handler
async def handle_a2a_task(task: A2ATask) -> A2AResponse:
# 1. Validate task
if not validate_task(task):
return A2AResponse(status="rejected")
# 2. Check payment (if x402 enabled)
if requires_payment(task):
if not verify_x402_payment(task):
return A2AResponse(
status="auth_required",
auth_type="x402"
)
# 3. Execute task
result = await execute_agent_task(task)
# 4. Return response
return A2AResponse(
status="completed",
result=result
)
# Discover A2A agents
agents = await a2a_client.discover(
query="research agent",
capabilities=["research", "analysis"]
)
for agent in agents:
print(f"{agent.name}: {agent.endpoint}")
print(f" Skills: {[s.name for s in agent.skills]}")
# Send task to external agent
task = await a2a_client.send_task(
agent_endpoint="https://external-agent.com/a2a",
skill="research",
input_data={
"query": "Latest developments in AI agents",
"depth": "comprehensive"
},
payment={ # Optional x402 payment
"amount": "0.10",
"asset": "USDC"
}
)
# Wait for completion
result = await task.wait_for_completion(timeout=300)
print(result.content)
# Stream task updates
async for update in a2a_client.stream_task(task_id):
if update.status == "working":
print(f"Progress: {update.progress}%")
elif update.type == "artifact":
print(f"Received: {update.artifact.name}")
elif update.status == "completed":
print(f"Result: {update.result}")
break
A2A + x402 enables paid agent collaboration:
┌─────────────────────────────────────────────────────────────────────┐
│ A2A + x402 FLOW │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ VIBE AGENT A2A + x402 EXTERNAL AGENT │
│ ┌─────────┐ ┌──────────┐ ┌─────────┐ │
│ │ │ 1.Discover │ │ │ │ │
│ │ Fetches │───────────▶│ Card │◀──────────│Publishes│ │
│ │ Card │ │ │ │ │ │
│ │ │ │ │ │ │ │
│ │ │ 2.Task │ │ │ │ │
│ │ Sends │───────────▶│ Requires │ │ │ │
│ │ Task │ │ Payment │ │ │ │
│ │ │◀───────────│ 402 │ │ │ │
│ │ │ │ │ │ │ │
│ │ │ 3.x402 Sig │ │ 4.Verify │ │ │
│ │ Pays │───────────▶│ Forward │──────────▶│ Execute │ │
│ │ │ │ │ │ │ │
│ │ │ 5.Result │ │ │ │ │
│ │ Receives│◀───────────│ │◀──────────│ Return │ │
│ └─────────┘ └──────────┘ └─────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘
class A2AClientTool(Tool):
"""Tool for agents to interact with A2A agents"""
@openapi_schema({
"name": "call_a2a_agent",
"description": "Send task to external A2A agent",
"parameters": {
"type": "object",
"properties": {
"agent_url": {"type": "string"},
"skill": {"type": "string"},
"input_data": {"type": "object"},
"max_payment": {"type": "string"}
}
}
})
async def call_a2a_agent(
self,
agent_url: str,
skill: str,
input_data: dict,
max_payment: str = "0.10"
) -> ToolResult:
try:
result = await self.a2a_client.send_task(
agent_endpoint=agent_url,
skill=skill,
input_data=input_data,
max_payment=max_payment
)
return self.success_response(result=result)
except A2AError as e:
return self.fail_response(str(e))
Next: MCP Integration →