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.

💻 Code Examples

A2A Server

Expose a Tyler agent via A2A

A2A Client

Connect to remote A2A agents

Multi-Agent

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:

# 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

from tyler.a2a import A2AAdapter

# Create adapter for connecting to remote agents
adapter = A2AAdapter()

Step 2: Connect to Remote Agents

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

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

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

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

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": "[e-mail]"
        },
        "vendor": "Tyler Framework"
    },
    authentication={
        "schemes": ["bearer"],
        "required": True
    }
)

Step 3: Start the A2A Server

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

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

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

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

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

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

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

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:

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):

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:

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

# 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

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

ImportError: a2a-sdk required

Problem: Getting import errors when using A2A features.

Solution: A2A SDK is included with Tyler. Ensure you have the latest version:

uv add slide-tyler --upgrade

Connection Refused Errors

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

# 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}")

Push Notification Failures

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

# 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")

Artifact Retrieval Issues

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

# 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

A2A Integration Best Practices

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:

A2A Concepts

Deep dive into A2A protocol concepts

Multi-Agent Patterns

Advanced agent delegation patterns

A2A Specification

Official A2A Protocol v0.3.0 specification

Production Deployment

Deploy A2A agents to production