Agent-to-Agent (A2A) protocol

Using A2A Protocol v0.3.0 to enable multi-agent coordination and delegation across platforms

What is A2A?

The Agent-to-Agent (A2A) Protocol is an open standard that enables AI agents from different platforms and frameworks to communicate and collaborate effectively. Slide’s Tyler package includes comprehensive A2A v0.3.0 support, allowing your agents to:

• Delegate tasks to remote A2A-compatible agents
• Expose Tyler agents as A2A endpoints for other systems
• Stream responses in real-time with token-level streaming via SSE
• Exchange files, structured data, and text between agents
• Produce and consume Artifacts as task deliverables
• Group related tasks with context IDs
• Receive real-time updates via push notifications
• Enable seamless interoperability between different agent frameworks

Use A2A for agent-to-agent delegation and interoperability. For general external tool or data-source integration, use MCP.

💻 Code Examples

A2A Server

Expose a Tyler agent via A2A

A2A Client

Connect to remote A2A agents

Multi-Agent

Coordinate multiple A2A agents

A2A Architecture in Slide

Quick start

Delegating to Remote A2A Agents

import asyncio
from tyler import Agent, Thread, Message
from tyler.a2a import A2AAdapter

async def main():
    # Create A2A adapter
    adapter = A2AAdapter()
    
    # Connect to remote A2A agents
    await adapter.connect("research_agent", "https://research-service.example.com")
    await adapter.connect("analysis_agent", "https://analysis-service.example.com")
    
    # Create Tyler agent with delegation capabilities
    agent = Agent(
        name="Project Coordinator",
        model_name="gpt-4.1",
        purpose="To coordinate complex projects using specialized remote agents",
        tools=adapter.get_tools_for_agent()  # Gets delegation tools
    )
    
    # Create a complex request that will be delegated
    thread = Thread()
    message = Message(
        role="user",
        content="Research quantum computing trends and analyze business opportunities"
    )
    thread.add_message(message)
    
    # Agent will automatically delegate to appropriate remote agents
    result = await agent.run(thread)

if __name__ == "__main__":
    asyncio.run(main())

Exposing Tyler Agents via A2A

import asyncio
from tyler import Agent
from tyler.a2a import A2AServer
from lye import WEB_TOOLS, FILES_TOOLS

async def main():
    # Create a Tyler agent with specific capabilities
    tyler_agent = Agent(
        name="Research Assistant",
        model_name="gpt-4.1",
        purpose="Advanced research specialist with web search and document processing",
        tools=[*WEB_TOOLS, *FILES_TOOLS]
    )
    
    # Create A2A server to expose the agent with authentication
    server = A2AServer(
        tyler_agent=tyler_agent,
        agent_card={
            "name": "Tyler Research Assistant",
            "description": "AI research specialist with web and document capabilities",
            "capabilities": ["web_research", "document_processing", "data_analysis"],
            "version": "1.0.0"
        },
        authentication={
            "schemes": ["bearer"],
            "required": True
        }
    )
    
    # Start the A2A server
    await server.start_server(host="0.0.0.0", port=8000)
    # Agent is now accessible at http://localhost:8000

if __name__ == "__main__":
    asyncio.run(main())

A2A Protocol v0.3.0 Features

Tyler fully supports the A2A Protocol v0.3.0 specification, including:

1. All Part Types

A2A messages can contain multiple types of content:

from tyler.a2a import TextPart, FilePart, DataPart

# Text content
text = TextPart(text="Analyze this document")

# File content (inline with bytes)
with open("document.pdf", "rb") as f:
    file_data = f.read()
file_part = FilePart(
    name="document.pdf", 
    media_type="application/pdf", 
    file_with_bytes=file_data
)

# File content (by URI reference)
remote_file = FilePart(
    name="report.xlsx", 
    media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
    file_with_uri="https://cdn.example.com/files/report.xlsx"
)

# Structured data
data_part = DataPart(data={
    "analysis_type": "financial",
    "parameters": {"period": "Q4", "metrics": ["revenue", "profit"]}
})

2. Artifacts

Artifacts are the formal deliverables of a task, providing structured outputs:

from tyler.a2a import Artifact, TextPart, DataPart

# Create an artifact with task results
artifact = Artifact.create(
    name="Market Analysis Report",
    parts=[
        TextPart(text="Executive Summary: Market growth projected at 15%..."),
        DataPart(data={"growth_rate": 0.15, "confidence": 0.92}),
    ],
    metadata={"analysis_version": "2.0"}
)

# Access artifact properties
print(f"Artifact ID: {artifact.artifact_id}")
print(f"Created at: {artifact.created_at}")

3. Context ID for Task Grouping

Group related tasks together using context IDs:

from tyler.a2a import A2AClient

client = A2AClient()
await client.connect("agent", "https://agent.example.com")

# Create multiple related tasks in the same context
context_id = "project-alpha-research"

task1_id = await client.create_task(
    "agent",
    "Research market trends",
    context_id=context_id
)

task2_id = await client.create_task(
    "agent", 
    "Analyze competitor landscape",
    context_id=context_id
)

# Get all tasks in a context
related_tasks = client.get_tasks_by_context(context_id)

4. Real-Time Token Streaming

A2A clients can receive response tokens as they’re generated via Server-Sent Events (SSE):

from tyler.a2a import A2AClient

client = A2AClient()
await client.connect("agent", "https://agent.example.com")

# Create a task
task_id = await client.create_task("agent", "Write a detailed analysis")

# Stream response tokens in real-time
async for message in client.stream_task_messages("agent", task_id):
    # Tokens arrive as they're generated by the LLM
    print(message.get("content", ""), end="", flush=True)

The A2A server uses Tyler’s streaming internally for all requests. The A2A SDK handles delivery:

• message/send: Returns aggregated response after completion
• message/stream: Streams tokens via SSE as they’re generated

For non-streaming requests, use send_task which waits for completion:

from tyler.a2a import A2AClient

client = A2AClient()
await client.connect("agent", "https://agent.example.com")

# Send task and wait for complete response (non-streaming)
result = await client.send_task("agent", "Summarize this document")

# Result contains the full response and artifacts
print(f"Status: {result.status}")
for artifact in result.artifacts:
    print(f"Artifact: {artifact.name}")
    for part in artifact.parts:
        print(part.text)

5. Push Notifications

Receive real-time updates for long-running tasks via webhooks:

from tyler.a2a import A2AClient, PushNotificationConfig

client = A2AClient()
await client.connect("agent", "https://agent.example.com")

# Configure push notifications
push_config = PushNotificationConfig(
    webhook_url="https://your-service.com/webhooks/a2a",
    events=["task.created", "task.updated", "task.completed", "task.artifact"],
    headers={"Authorization": "Bearer your-webhook-token"},
    secret="your-hmac-secret"  # For signature verification
)

# Create task with push notifications
task_id = await client.create_task(
    "agent",
    "Generate comprehensive market report",
    push_notification_config=push_config
)

# Your webhook will receive events like:
# {
#   "event_type": "task.completed",
#   "task_id": "...",
#   "timestamp": "...",
#   "data": {"status": "completed", "artifacts": [...]}
# }

6. Authentication Declaration

Declare authentication requirements in your Agent Card:

from tyler.a2a import A2AServer

server = A2AServer(
    tyler_agent=my_agent,
    authentication={
        "schemes": ["bearer", "api_key"],
        "required": True
    }
)

A2A Protocol Features

Agent Discovery

A2A uses Agent Cards to describe agent capabilities:

agent_card = {
    "name": "Tyler Research Assistant",
    "version": "1.0.0", 
    "description": "Specialized research agent with web search capabilities",
    "protocol_version": "0.3.0",
    "capabilities": [
        "web_research",
        "fact_checking", 
        "data_analysis",
        "document_processing",
        "artifacts"
    ],
    "push_notifications": {
        "supported": True,
        "events": ["task.created", "task.updated", "task.completed", "task.artifact"]
    },
    "authentication": {
        "schemes": ["bearer"],
        "required": False
    },
    "contact": {
        "name": "Your Organization",
        "email": "[e-mail]"
    }
}

Task Management

A2A provides full task lifecycle management:

# Create and track tasks
task_id = await adapter.client.create_task("research_agent", "Analyze market trends")

# Get task status
status = await adapter.client.get_task_status("research_agent", task_id)
print(f"Task status: {status['status']}")
print(f"Has artifacts: {status['has_artifacts']}")

# Get task artifacts
artifacts = await adapter.client.get_task_artifacts("research_agent", task_id)
for artifact in artifacts:
    print(f"Artifact: {artifact.name}")
    for part in artifact.parts:
        if isinstance(part, TextPart):
            print(f"  Text: {part.text[:100]}...")

# Stream real-time responses
async for message in adapter.client.stream_task_messages("research_agent", task_id):
    print(f"Agent response: {message['content']}")

Secure Communication

A2A supports standard web security practices:

# Connect with authentication
await adapter.connect(
    "secure_agent",
    "https://secure-service.example.com",
    headers={"Authorization": "Bearer your-token"}
)

Multi-Agent Coordination

Creating Specialized Agent Networks

import asyncio
from tyler import Agent
from tyler.a2a import A2AAdapter, A2AServer
from lye import WEB_TOOLS, FILES_TOOLS

async def create_agent_network():
    """Create a network of specialized A2A agents."""
    
    # Create specialized agents
    research_agent = Agent(
        name="Research Specialist",
        tools=[*WEB_TOOLS],
        purpose="Web research and information gathering expert"
    )
    
    analysis_agent = Agent(
        name="Analysis Specialist", 
        tools=[*FILES_TOOLS],
        purpose="Data analysis and strategic insights expert"
    )
    
    # Expose agents as A2A servers
    research_server = A2AServer(research_agent)
    analysis_server = A2AServer(analysis_agent)
    
    # Start servers (in production, these would be separate services)
    await research_server.start_server(port=8001)
    await analysis_server.start_server(port=8002)
    
    # Create coordinator that uses both agents
    adapter = A2AAdapter()
    await adapter.connect("research", "http://localhost:8001")
    await adapter.connect("analysis", "http://localhost:8002")
    
    coordinator = Agent(
        name="Project Coordinator",
        purpose="Orchestrate complex projects using specialized agents",
        tools=adapter.get_tools_for_agent()
    )
    
    return coordinator

Task Delegation with Files and Data

from tyler.a2a import A2AAdapter, FilePart, DataPart

adapter = A2AAdapter()
await adapter.connect("analysis_agent", "https://analysis.example.com")

# Create task with file attachment
task_id = await adapter.create_task_with_files(
    "analysis_agent",
    "Analyze the attached financial data",
    files=["./data/q4_report.xlsx", "./data/projections.csv"],
    context_id="financial-analysis-2024"
)

# Create task with structured data
task_id = await adapter.create_task_with_data(
    "analysis_agent",
    "Run analysis with these parameters",
    data={
        "analysis_type": "trend",
        "time_range": {"start": "2024-01", "end": "2024-12"},
        "metrics": ["revenue", "growth", "churn"]
    }
)

A2A vs MCP Comparison

Feature	A2A Protocol	MCP Protocol
Purpose	Agent-to-agent communication	Model-to-tool communication
Scope	Multi-agent coordination	Tool integration
Communication	Task delegation & responses	Tool calls & results
Architecture	Agent ↔ Agent	Model ↔ Tools
Use Cases	Distributed agent systems	Tool ecosystem integration

Production Deployment

Server Configuration

# Production A2A server setup
server = A2AServer(
    tyler_agent=production_agent,
    agent_card={
        "name": "Production Research Agent",
        "version": "2.0.0",
        "description": "Enterprise research capabilities",
        "capabilities": ["web_research", "document_analysis", "market_intelligence", "artifacts"],
        "contact": {"email": "[e-mail]"}
    },
    authentication={
        "schemes": ["bearer"],
        "required": True
    }
)

# Start with production settings
await server.start_server(
    host="0.0.0.0",
    port=8000,
)

Installation and Setup

Install A2A Dependencies

# Install Tyler (includes A2A support)
uv add slide-tyler

# A2A SDK is included as a dependency

Environment Setup

# Optional: Configure A2A settings
import os

# Set A2A agent base URL for discovery
os.environ["A2A_BASE_URL"] = "https://your-agent-service.com"

# Configure authentication if needed
os.environ["A2A_AUTH_TOKEN"] = "your-auth-token"

Security Considerations

Production Security: Always use HTTPS endpoints, implement proper authentication, and validate agent cards in production environments.

Authentication Methods

# Token-based authentication
await adapter.connect(
    "secure_agent",
    "https://secure-agent.example.com",
    headers={"Authorization": "Bearer your-secure-token"}
)

# Custom headers for API keys
await adapter.connect(
    "api_agent", 
    "https://api-agent.example.com",
    headers={
        "X-API-Key": "your-api-key",
        "X-Client-Version": "1.0.0"
    }
)

Webhook Security

from tyler.a2a import PushNotificationConfig

# Configure secure webhooks with HMAC signing
push_config = PushNotificationConfig(
    webhook_url="https://your-service.com/webhooks/a2a",
    secret="your-hmac-secret",  # Webhook payloads will be signed
    events=["task.completed", "task.artifact"]
)

# Verify webhook signature in your handler
import hmac
import hashlib

def verify_webhook(payload: str, signature: str, secret: str) -> bool:
    expected = hmac.new(
        secret.encode(),
        payload.encode(),
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(f"sha256={expected}", signature)

Best practices

Design Principles

• Specialized Agents: Create agents with focused, well-defined capabilities
• Clear Interfaces: Use descriptive agent cards and task descriptions
• Use Artifacts: Structure task outputs as formal Artifacts for better interoperability
• Context Grouping: Use context IDs to group related tasks for better tracking
• Push Notifications: Use webhooks for long-running tasks instead of polling
• Error Handling: Implement robust connection and task failure handling
• Monitoring: Track agent health and task completion rates
• Security: Always use secure connections and proper authentication

Next steps

A2A Integration Guide

Step-by-step A2A integration tutorial

Multi-Agent Patterns

Learn advanced delegation patterns

A2A Specification

Read the official A2A protocol specification

Agent Architecture

Understand Tyler’s agent architecture