> ## Documentation Index
> Fetch the complete documentation index at: https://slide.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.
# A2A integration guide
> Step-by-step guide to integrating Agent-to-Agent (A2A) Protocol v0.3.0 with Tyler agents
This guide walks you through integrating the Agent-to-Agent (A2A) Protocol v0.3.0 with your Tyler agents, enabling multi-agent coordination and delegation across different platforms.
Use A2A when one agent delegates to or exposes another agent. Use [MCP](/guides/mcp-integration) for external tools, services, and data sources.
**💻 Code Examples**
Expose a Tyler agent via A2A
Connect to remote A2A agents
Coordinate multiple A2A agents
## Prerequisites
Before starting, ensure you have:
* Tyler installed and working
* Basic understanding of Tyler agents
* Python 3.11+ environment
## Installation
Tyler includes A2A support out of the box:
```bash theme={null}
# Install Tyler (includes a2a-sdk)
uv add slide-tyler
# For server functionality, these are already included:
# - fastapi
# - uvicorn
```
The A2A integration is fully supported in Tyler. All dependencies are included automatically.
## Part 1: Connecting to Remote A2A Agents (Client Mode)
### Step 1: Create an A2A Adapter
```python theme={null}
from tyler.a2a import A2AAdapter
# Create adapter for connecting to remote agents
adapter = A2AAdapter()
```
### Step 2: Connect to Remote Agents
```python theme={null}
import asyncio
async def connect_to_agents():
# Connect to a research specialist agent
research_connected = await adapter.connect(
name="research_agent",
base_url="https://research-ai.example.com"
)
# Connect to an analysis specialist agent
analysis_connected = await adapter.connect(
name="analysis_agent",
base_url="https://analysis-ai.example.com"
)
if research_connected and analysis_connected:
print("Connected to both remote agents")
# Check agent capabilities
for agent_name in ["research_agent", "analysis_agent"]:
info = adapter.client.get_connection_info(agent_name)
print(f"{agent_name}:")
print(f" Protocol version: {info['protocol_version']}")
print(f" Capabilities: {info['capabilities']}")
print(f" Push notifications: {info['push_notifications_supported']}")
else:
print("Failed to connect to one or more agents")
asyncio.run(connect_to_agents())
```
### Step 3: Create Tyler Agent with Delegation Tools
```python theme={null}
from tyler import Agent
# Get delegation tools from connected agents
delegation_tools = adapter.get_tools_for_agent()
# Create Tyler agent that can delegate tasks
coordinator = Agent(
name="Project Coordinator",
model_name="gpt-4.1",
purpose="""You coordinate complex projects by delegating specialized tasks to remote agents.
You have access to:
- Research agent: For web research, fact-checking, and information gathering
- Analysis agent: For data analysis, insights, and strategic recommendations
Use these agents strategically to break down complex requests.""",
tools=delegation_tools
)
```
### Step 4: Use the Coordinating Agent
```python theme={null}
from tyler import Thread, Message
async def coordinate_project():
# Create a complex request
thread = Thread()
thread.add_message(Message(
role="user",
content="""I need a comprehensive market analysis for electric vehicle charging stations.
Please:
1. Research current market size, key players, and growth trends
2. Analyze competitive landscape and identify opportunities
3. Provide strategic recommendations for market entry
"""
))
# The coordinator will automatically delegate to appropriate agents
result = await coordinator.run(thread)
# Print the coordinated response
for message in result.thread.messages:
if message.role == "assistant":
print(f"Coordinator: {message.content}")
asyncio.run(coordinate_project())
```
## Part 2: Exposing Tyler Agents via A2A (Server Mode)
### Step 1: Create a Specialized Tyler Agent
```python theme={null}
from tyler import Agent
from lye import WEB_TOOLS, FILES_TOOLS
# Create a specialized research agent
research_agent = Agent(
name="Research Specialist",
model_name="gpt-4.1",
purpose="""You are an expert research specialist with web search and document processing capabilities.
Your expertise includes:
- Comprehensive web research and fact-finding
- Academic and market research
- Document analysis and summarization
- Competitive intelligence gathering
Always provide well-sourced, accurate information.""",
tools=[*WEB_TOOLS, *FILES_TOOLS]
)
```
### Step 2: Create A2A Server with Authentication
```python theme={null}
from tyler.a2a import A2AServer
# Create server to expose the agent
server = A2AServer(
tyler_agent=research_agent,
agent_card={
"name": "Tyler Research Specialist",
"version": "1.0.0",
"description": "AI research specialist with web search and document processing",
"capabilities": [
"web_research",
"fact_checking",
"document_analysis",
"market_research",
"competitive_intelligence",
"artifacts"
],
"contact": {
"name": "Your Organization",
"email": "ai-support@yourorg.com"
},
"vendor": "Tyler Framework"
},
authentication={
"schemes": ["bearer"],
"required": True
}
)
```
### Step 3: Start the A2A Server
```python theme={null}
async def start_research_service():
print("Starting Tyler Research Specialist A2A Server...")
print("Other agents can connect at: http://localhost:8000")
print("Agent Card available at: http://localhost:8000/.well-known/agent-card.json")
# Start the server (this will run indefinitely)
await server.start_server(host="0.0.0.0", port=8000)
# Run the server
if __name__ == "__main__":
try:
asyncio.run(start_research_service())
except KeyboardInterrupt:
print("\nServer stopped by user")
```
## Part 3: Working with Part Types
### Sending Files to Remote Agents
```python theme={null}
from tyler.a2a import A2AAdapter, FilePart
adapter = A2AAdapter()
await adapter.connect("document_processor", "https://docs.example.com")
# Send a file for processing
task_id = await adapter.create_task_with_files(
"document_processor",
"Summarize the key points from this document",
files=["./reports/annual_report_2024.pdf"],
context_id="annual-review"
)
# Or create FilePart manually for more control
file_part = FilePart.from_path("./data/analysis.xlsx")
print(f"File: {file_part.name}, Size: {len(file_part.file_with_bytes)} bytes")
```
### Sending Structured Data
```python theme={null}
from tyler.a2a import A2AAdapter, DataPart
adapter = A2AAdapter()
await adapter.connect("analysis_agent", "https://analysis.example.com")
# Send structured data for analysis
task_id = await adapter.create_task_with_data(
"analysis_agent",
"Perform trend analysis on this data",
data={
"dataset": "sales_2024",
"metrics": ["revenue", "units", "margin"],
"grouping": "monthly",
"filters": {
"region": ["NA", "EU"],
"product_category": "electronics"
}
},
context_id="sales-analysis"
)
```
## Part 4: Working with Artifacts
### Retrieving Task Artifacts
```python theme={null}
from tyler.a2a import A2AClient, TextPart, DataPart
client = A2AClient()
await client.connect("agent", "https://agent.example.com")
# Create a task
task_id = await client.create_task("agent", "Generate a comprehensive report")
# Wait for completion (in production, use push notifications instead)
import asyncio
while True:
status = await client.get_task_status("agent", task_id)
if status["status"] in ["completed", "error"]:
break
await asyncio.sleep(1)
# Get artifacts
artifacts = await client.get_task_artifacts("agent", task_id)
for artifact in artifacts:
print(f"Artifact: {artifact.name}")
print(f" ID: {artifact.artifact_id}")
print(f" Created: {artifact.created_at}")
for part in artifact.parts:
if isinstance(part, TextPart):
print(f" Text content: {part.text[:200]}...")
elif isinstance(part, DataPart):
print(f" Data: {part.data}")
```
## Part 5: Context-Based Task Grouping
```python theme={null}
from tyler.a2a import A2AClient
client = A2AClient()
await client.connect("agent", "https://agent.example.com")
# Define a context for related tasks
context_id = "market-research-project-q4"
# Create multiple related tasks
tasks = [
("Research competitor pricing", "phase1-research"),
("Analyze market trends", "phase2-analysis"),
("Generate recommendations", "phase3-synthesis"),
]
task_ids = []
for description, tag in tasks:
task_id = await client.create_task(
"agent",
f"[{tag}] {description}",
context_id=context_id
)
task_ids.append(task_id)
# Get all tasks in the context
related_task_ids = client.get_tasks_by_context(context_id)
print(f"Tasks in context: {len(related_task_ids)}")
```
## Part 6: Push Notifications for Long-Running Tasks
### Configuring Push Notifications
```python theme={null}
from tyler.a2a import A2AClient, PushNotificationConfig
client = A2AClient()
await client.connect("agent", "https://agent.example.com")
# Configure webhook for notifications
push_config = PushNotificationConfig(
webhook_url="https://your-app.com/webhooks/a2a-events",
events=[
"task.created",
"task.updated",
"task.completed",
"task.failed",
"task.artifact"
],
headers={
"Authorization": "Bearer your-webhook-secret",
"X-App-ID": "my-coordinator"
},
secret="hmac-signing-secret" # For payload verification
)
# Create task with push notifications
task_id = await client.create_task(
"agent",
"Perform comprehensive analysis (this may take a while)",
push_notification_config=push_config
)
print(f"Task {task_id} created. Updates will be sent to your webhook.")
```
### Handling Webhook Events
```python theme={null}
from fastapi import FastAPI, Request, HTTPException
import hmac
import hashlib
app = FastAPI()
WEBHOOK_SECRET = "hmac-signing-secret"
@app.post("/webhooks/a2a-events")
async def handle_a2a_event(request: Request):
# Verify signature
signature = request.headers.get("X-A2A-Signature", "")
body = await request.body()
expected = hmac.new(
WEBHOOK_SECRET.encode(),
body,
hashlib.sha256
).hexdigest()
if not hmac.compare_digest(f"sha256={expected}", signature):
raise HTTPException(status_code=401, detail="Invalid signature")
# Process event
event = await request.json()
event_type = event["event_type"]
task_id = event["task_id"]
if event_type == "task.created":
print(f"Task {task_id} created")
elif event_type == "task.updated":
print(f"Task {task_id} status: {event['data']['status']}")
elif event_type == "task.completed":
print(f"Task {task_id} completed!")
artifacts = event["data"].get("artifacts", [])
print(f" Produced {len(artifacts)} artifacts")
elif event_type == "task.failed":
print(f"Task {task_id} failed: {event['data']['error']}")
elif event_type == "task.artifact":
artifact = event["data"]["artifact"]
print(f"New artifact: {artifact['name']}")
return {"status": "received"}
```
## Part 7: Advanced Multi-Agent Patterns
### Creating Agent Networks
```python theme={null}
import asyncio
from typing import Dict, Any
from tyler.a2a import A2AAdapter
class AgentNetwork:
"""Manages a network of specialized A2A agents."""
def __init__(self):
self.adapter = A2AAdapter()
self.agents: Dict[str, Any] = {}
async def add_agent(self, name: str, base_url: str, specialization: str):
"""Add a specialized agent to the network."""
connected = await self.adapter.connect(name, base_url)
if connected:
info = self.adapter.client.get_connection_info(name)
self.agents[name] = {
"base_url": base_url,
"specialization": specialization,
"status": "connected",
"protocol_version": info.get("protocol_version"),
"capabilities": info.get("capabilities", []),
}
print(f"Added {name} ({specialization})")
else:
print(f"Failed to connect to {name}")
async def create_coordinator(self) -> "Agent":
"""Create a coordinator agent for this network."""
from tyler import Agent
tools = self.adapter.get_tools_for_agent()
# Build purpose description based on connected agents
specializations = [agent["specialization"] for agent in self.agents.values()]
purpose = f"""You coordinate complex tasks across a network of specialized agents.
Available specialists: {', '.join(specializations)}
Break down complex requests and delegate appropriately to leverage each agent's expertise."""
return Agent(
name="Network Coordinator",
model_name="gpt-4.1",
purpose=purpose,
tools=tools
)
async def health_check(self):
"""Check the health of all connected agents."""
for name, info in self.agents.items():
try:
status = await self.adapter.get_agent_status(name)
if status:
print(f"{name}: healthy ({status.get('active_tasks', 0)} active tasks)")
else:
print(f"{name}: unreachable")
except Exception as e:
print(f"{name}: error - {e}")
# Example usage
async def setup_enterprise_network():
network = AgentNetwork()
# Add specialized agents
await network.add_agent("research", "https://research.corp.com", "Research & Intelligence")
await network.add_agent("analysis", "https://analytics.corp.com", "Data Analysis")
await network.add_agent("compliance", "https://compliance.corp.com", "Legal & Compliance")
# Create coordinator
coordinator = await network.create_coordinator()
# Health check
await network.health_check()
return coordinator, network
```
### Task Streaming and Monitoring
Tyler's A2A server supports both streaming and non-streaming request modes:
#### Non-Streaming (message/send)
For simple requests where you want to wait for the complete response:
```python theme={null}
from tyler.a2a import A2AClient
async def send_remote_task():
"""Send a task and wait for complete response."""
client = A2AClient()
await client.connect("agent", "https://agent.example.com")
# Send task and wait for full response
result = await client.send_task(
"agent",
"Create a brief executive summary"
)
# Access the complete response
print(f"Status: {result.status}")
for artifact in result.artifacts:
for part in artifact.parts:
print(part.text)
```
#### Real-Time Streaming (message/stream)
When clients call `message/stream`, they receive response tokens as they're generated by the LLM via Server-Sent Events (SSE):
```python theme={null}
from tyler.a2a import A2AClient
async def stream_remote_task():
"""Stream responses from a remote A2A agent."""
client = A2AClient()
await client.connect("agent", "https://agent.example.com")
# Create a task
task_id = await client.create_task(
"agent",
"Create a comprehensive business plan for a new AI startup"
)
print("Streaming response...")
print("=" * 50)
# Stream tokens as they arrive
async for message in client.stream_task_messages("agent", task_id):
content = message.get("content", "")
print(content, end="", flush=True)
print("\n\nTask complete!")
```
For local coordinating agents, you can also stream using Tyler's native streaming:
```python theme={null}
from tyler.models.execution import EventType
async def stream_coordinated_task():
"""Example of streaming responses from coordinated agents."""
thread = Thread()
thread.add_message(Message(
role="user",
content="Create a comprehensive business plan for a new AI startup"
))
print("Starting coordinated task execution...")
print("=" * 50)
async for update in coordinator.stream(thread):
if update.type == EventType.LLM_STREAM_CHUNK:
print(update.data.get("content_chunk", ""), end="", flush=True)
elif update.type == EventType.TOOL_SELECTED:
tool_name = update.data.get("tool_name", "")
if "delegate_to_" in tool_name:
agent_name = tool_name.replace("delegate_to_", "")
print(f"\n\nDelegating to {agent_name}...")
print()
elif update.type == EventType.EXECUTION_COMPLETE:
print("\n\nTask coordination complete!")
```
## Part 8: Production Considerations
### Security and Authentication
```python theme={null}
# Secure A2A connections
async def secure_connection():
adapter = A2AAdapter()
# Connect with authentication
await adapter.connect(
name="secure_agent",
base_url="https://secure-agents.company.com",
headers={
"Authorization": "Bearer your-secure-token",
"X-API-Version": "1.0"
}
)
```
### Error Handling and Resilience
```python theme={null}
from typing import Optional
class ResilientA2AAdapter:
"""A2A adapter with built-in resilience patterns."""
def __init__(self):
self.adapter = A2AAdapter()
self.fallback_agents: Dict[str, str] = {}
async def connect_with_fallback(
self,
primary_url: str,
fallback_url: Optional[str] = None
):
"""Connect with automatic fallback."""
try:
success = await self.adapter.connect("primary", primary_url)
if success:
return True
except Exception as e:
print(f"Primary connection failed: {e}")
if fallback_url:
try:
return await self.adapter.connect("fallback", fallback_url)
except Exception as e:
print(f"Fallback connection failed: {e}")
return False
async def robust_delegation(self, task: str, max_retries: int = 3):
"""Delegate with retry logic."""
for attempt in range(max_retries):
try:
return await self._execute_task(task)
except Exception as e:
if attempt == max_retries - 1:
raise
print(f"Attempt {attempt + 1} failed: {e}. Retrying...")
await asyncio.sleep(2 ** attempt) # Exponential backoff
```
## Troubleshooting
### Common Issues
**Problem**: Getting import errors when using A2A features.
**Solution**: A2A SDK is included with Tyler. Ensure you have the latest version:
```bash theme={null}
uv add slide-tyler --upgrade
```
**Problem**: Cannot connect to remote A2A agents.
**Solutions**:
* Verify the agent URL is correct and accessible
* Check network connectivity and firewall settings
* Ensure the remote agent is running and healthy
* Verify authentication credentials if required
```python theme={null}
# Test connection manually
try:
connected = await adapter.connect("test", "https://agent.example.com")
if connected:
info = adapter.client.get_connection_info("test")
print(f"Connected! Protocol: {info['protocol_version']}")
else:
print("Connection failed - check agent status")
except Exception as e:
print(f"Connection error: {e}")
```
**Problem**: Webhook notifications are not being received.
**Solutions**:
* Verify webhook URL is accessible from the agent's network
* Check that the URL uses HTTPS (required for security)
* Ensure webhook server returns 2xx status codes
* Verify HMAC signature validation if using secrets
```python theme={null}
# Debug webhook configuration
from tyler.a2a import validate_webhook_url
url = "https://your-service.com/webhook"
if validate_webhook_url(url):
print("URL is valid")
else:
print("URL validation failed - check HTTPS and accessibility")
```
**Problem**: Cannot retrieve artifacts from completed tasks.
**Solutions**:
* Ensure task has actually completed
* Check that the remote agent supports artifacts
* Verify connection is still active
```python theme={null}
# Check task status before retrieving artifacts
status = await client.get_task_status("agent", task_id)
print(f"Status: {status['status']}")
print(f"Has artifacts: {status['has_artifacts']}")
if status["status"] == "completed" and status["has_artifacts"]:
artifacts = await client.get_task_artifacts("agent", task_id)
```
## Best Practices Summary
**Connection Management**
* Always handle connection failures gracefully
* Implement health checks for connected agents
* Use connection pooling for high-throughput scenarios
**Task Delegation**
* Be specific in task descriptions for remote agents
* Use context IDs to group related tasks
* Monitor task progress with push notifications for long-running tasks
* Retrieve and process artifacts for structured results
**Security**
* Always use HTTPS in production
* Implement proper authentication and authorization
* Use HMAC signing for webhook verification
* Validate agent cards and capabilities
**Performance**
* Monitor delegation latency and success rates
* Use streaming for long-running tasks
* Implement caching where appropriate
**Error Handling**
* Plan for network failures and agent unavailability
* Log delegation attempts and outcomes
* Provide meaningful error messages to users
## Next steps
Now that you have A2A integration working, explore these advanced topics:
Deep dive into A2A protocol concepts
Advanced agent delegation patterns
Official A2A Protocol v0.3.0 specification
Deploy A2A agents to production