Search⌘ K
AI Features
Home
Courses
Introduction to Google’s Agent2Agent (A2A) Protocol

Using A2A SDK

Explore how to upgrade from raw A2A implementations to using the official A2A SDK for agent development. Understand core SDK components like AgentExecutor, DefaultRequestHandler, and task lifecycle management. Learn to build scalable, maintainable agents with enhanced protocol handling and production features.

We'll cover the following...

We built a complete A2A agent from scratch using FastAPI and raw JSON-RPC handling. That gave us a deep understanding of how A2A works behind the scenes: discovery endpoints, message structures, task life cycles, and protocol mechanics. Now, we’ll see how the official A2A SDK transforms that functionality into something more elegant, maintainable, and production-ready.

Think of our previous implementation as learning to drive a manual transmission; we learned exactly how the engine, clutch, and gears work together. Now, we’re upgrading to an automatic transmission that handles those details for you, so you can focus on where you want to go.

We’ll recreate the same echo agent, but this time, by using the official A2A Python SDK. The functionality will be identical: receive messages and echo them back, but you’ll see how much of the complexity is eliminated by the SDK. More importantly, this SDK version is ready for production with built-in error handling, observability, and extensibility.

What do we need to know before using the A2A SDK?

Before exploring the code, let’s understand the key components that make the A2A SDK powerful.

  • AgentExecutor: Where your business logic lives. Instead of manually parsing JSON-RPC requests and constructing task objects, you implement execute() and cancel() methods that receive validated context and publish events.

  • DefaultRequestHandler: Coordinates the entire request life cycle. It manages task creation, validates requests, calls your AgentExecutor, handles streaming, and constructs proper A2A responses.

  • A2AFastAPIApplication: Replaces your manual FastAPI setup. It manages all A2A protocol details, including discovery endpoints, JSON-RPC routing, validation, error handling, and response formatting.

  • InMemoryTaskStore: Manages task persistence and state. The SDK requires this to track task life cycles and history.

  • Event queue: Used by your AgentExecutor to publish events, communicating results to clients. The SDK handles all the JSON-RPC wrapping and task construction.

The beauty of this architecture lies in its separation of concerns: you focus on what your agent does (business logic), while the SDK handles how it communicates (protocol implementation).

How to build an A2A SDK-powered server

As previously mentioned, we’ll build our agent incrementally so you can see exactly what each part does. Let’s create our new agent in a file called sdk_echo_agent.py and add code step-by-step. Let’s start with the essential imports and understand what each one provides:

# Core A2A types for defining agent capabilities
from a2a.types import AgentCard, AgentSkill, AgentCapabilities
# Server framework components
from a2a.server.apps.jsonrpc.fastapi_app import A2AFastAPIApplication
from a2a.server.request_handlers.default_request_handler import DefaultRequestHandler
from a2a.server.agent_execution.agent_executor import AgentExecutor
from a2a.server.tasks.inmemory_task_store import InMemoryTaskStore
# For running the server
import uvicorn
import asyncio
A2A SDK imports

  • Lines 1-2: Import the core A2A types for defining agent capabilities and metadata.

  • Lines 4-8: Import the real SDK components for building agents.

  • Lines 11-12: Import uvicorn and asyncio for running the HTTP server and handling async operations.

The SDK provides clean, typed interfaces with real classes from the actual codebase! Now, let’s define our agent using the SDK’s actual type-safe classes with correct field names:

# Add these agent definitions after the imports
# Define the agent's skill using real SDK classes
echo_skill = AgentSkill(
    id="echo_messages",
    name="Echo Messages", 
    description="Repeats whatever you say back to you",
    tags=["echo", "simple", "demo"],
    examples=[
        "Hello there!",
        "How are you doing?", 
        "Echo this message back to me"
    ],
    input_modes=["text/plain"],
    output_modes=["text/plain"]
)
# Define agent capabilities
capabilities = AgentCapabilities(
    streaming=False,
    push_notifications=False,
    state_transition_history=True
)
# Create the agent card using real SDK classes
agent_card = AgentCard(
    name="SDK Echo Agent",
    description="An echo agent built with the official A2A SDK to demonstrate best practices",
    url="http://localhost:8000",
    version="1.0.0",
    protocol_version="0.3.0",
    skills=[echo_skill],
    default_input_modes=["text/plain"],
    default_output_modes=["text/plain"],
    capabilities=capabilities
)
AgentSkill and AgentCard definition using A2A SDK

  • Lines 4–16: Define the agent skill with correct field names (input_modes, output_modes use underscores).

  • Lines 18–22: Create the AgentCapabilities object with the correct field names (push_notifications, state_transition_history).

  • Lines 25–36: Create the AgentCard with the correct field names (default_input_modes, default_output_modes, protocol_version).

Now for the core functionality, using the real AgentExecutor interface:

Python 3.10.4
# Add the executor class after the agent card
class EchoAgentExecutor(AgentExecutor):
    """
    Business logic implementation using the real A2A SDK AgentExecutor.
    The SDK handles all protocol complexity for us.
    """
    
    async def execute(self, context, event_queue):
        """
        Execute the echo logic using the real SDK pattern.
        
        Args:
            context: RequestContext containing the user's input and metadata
            event_queue: EventQueue for publishing response events
        """
        try:
            # Get the user's input - this returns a string with all text parts combined
            user_text = context.get_user_input()
            
            # Handle the case where no text was found
            if not user_text or user_text.strip() == "":
                # Create an error message using the SDK utility
                error_message = new_agent_text_message(
                    "I didn't receive any text to echo. Please send me a message with text content."
                )
                await event_queue.enqueue_event(error_message)
                return
            
            # Create and enqueue our echo response using the SDK utility
            echo_message = new_agent_text_message(f"You said: '{user_text.strip()}'")
            await event_queue.enqueue_event(echo_message)
            
        except Exception as e:
            # The SDK provides structured error handling
            error_message = new_agent_text_message(f"Error processing your message: {str(e)}")
            await event_queue.enqueue_event(error_message)
    
    async def cancel(self, task_id, event_queue):
        """
        Handle task cancellation using the real SDK interface.
        
        Args:
            task_id: The ID of the task being canceled
            event_queue: EventQueue for publishing cancellation events
        """
        cancel_message = new_agent_text_message(f"Echo task {task_id} has been canceled.")
        await event_queue.enqueue_event(cancel_message)

  • Lines 4–8: Define the AgentExecutor subclass using the real SDK base class.

  • Lines 10–16: Implement the required execute() method with the correct signature.

  • Lines 18–19: Get user input from context—get_user_input() returns a string, not a message object.

  • Lines 21–28: Handle the edge case and create an error message using new_agent_text_message().

  • Lines 30–32: Create the echo response and enqueue it with enqueue_event().

  • Lines 34–37: Handle exceptions with structured error messages.

  • Lines 39–47: Implement the required cancel() method for task cancellation.

This uses the real AgentExecutor interface with the correct method names and SDK structure.

Finally, wire everything together using the actual A2A SDK classes:

Python 3.10.4
# Add the application setup after the executor class
def create_app():
    """Create and configure the A2A application using real SDK components"""
    
    # Create the agent executor
    executor = EchoAgentExecutor()
    
    # Create task store for managing task state
    task_store = InMemoryTaskStore()
    
    # Create the request handler that coordinates everything
    request_handler = DefaultRequestHandler(
        agent_executor=executor,
        task_store=task_store
    )
    
    # Create the A2A FastAPI application
    app = A2AFastAPIApplication(
        agent_card=agent_card,
        http_handler=request_handler
    )
    
    # Build and return the configured FastAPI app
    return app.build()
# Complete the main section
if __name__ == "__main__":
    print("🚀 Starting A2A SDK Echo Agent...")
    print("📡 Agent Card: http://localhost:8000/.well-known/agent-card.json")
    print("🔗 A2A Endpoint: http://localhost:8000/")
    
    app = create_app()
    
    uvicorn.run(
        app,
        host="0.0.0.0",
        port=8000,
        log_level="info"
    )

  • Lines 4–7: Create the custom AgentExecutor instance.

  • Lines 9–10: Create the InMemoryTaskStore, required by DefaultRequestHandler.

  • Lines 12–16: Create the DefaultRequestHandler with the executor and task store.

  • Lines 18–22: Create the A2AFastAPIApplication with the Agent Card and request handler.

  • Lines 24–25: Build the FastAPI application using the .build() method.

  • Lines 29–31: Print startup information and create the configured application.

  • Lines 33–38: Start the uvicorn server with the proper configuration.

How does the SDK control flow work? Every client request enters the A2AFastAPIApplication, which routes it through the DefaultRequestHandler. The handler validates the request, manages task state using the InMemoryTaskStore, and calls your AgentExecutor.execute() method. Your executor enqueues results or progress events through the event queue, which the SDK automatically formats and streams back to the client.

The SDK automatically handles:

  • Serving the agent card at /.well-known/agent-card.json.

  • JSON-RPC 2.0 request parsing and validation.

  • Task life cycle management and unique ID generation.

  • Event queue management and response construction.

  • Error handling and response formatting.

Here’s the complete sdk_echo_agent.py using the real A2A SDK. Let’s start our new agent and test it. Press the “Run” button to see it communicating with the same client that we built in the previous lesson.

# Core A2A types for defining agent capabilities
from a2a.types import AgentCard, AgentSkill, AgentCapabilities

# Server framework components
from a2a.server.apps.jsonrpc.fastapi_app import A2AFastAPIApplication
from a2a.server.request_handlers.default_request_handler import DefaultRequestHandler
from a2a.server.agent_execution.agent_executor import AgentExecutor
from a2a.server.tasks.inmemory_task_store import InMemoryTaskStore

# Message utilities
from a2a.utils.message import new_agent_text_message

# For running the server
import uvicorn

# Define the agent's skill using real SDK classes
echo_skill = AgentSkill(
    id="echo_messages",
    name="Echo Messages",
    description="Repeats whatever you say back to you.",
    tags=["echo", "simple", "demo"],
    examples=[
        "Hello there!",
        "How are you doing?",
        "Echo this message back to me."
    ],
    input_modes=["text/plain"],
    output_modes=["text/plain"]
)

# Define agent capabilities
capabilities = AgentCapabilities(
    streaming=False,
    push_notifications=False,
    state_transition_history=True
)

# Create the agent card using real SDK classes
agent_card = AgentCard(
    name="SDK Echo Agent",
    description="An echo agent built with the official A2A SDK to demonstrate best practices.",
    url="http://localhost:8000",
    version="1.0.0",
    protocol_version="0.3.0",
    skills=[echo_skill],
    default_input_modes=["text/plain"],
    default_output_modes=["text/plain"],
    capabilities=capabilities
)

class EchoAgentExecutor(AgentExecutor):
    """
    Business logic implementation using the real A2A SDK AgentExecutor.
    The SDK handles all protocol complexity for us.
    """
    
    async def execute(self, context, event_queue):
        """
        Execute the echo logic using the SDK pattern.
        
        Args:
            context: RequestContext containing the user's input and metadata.
            event_queue: EventQueue for publishing response events.
        """
        try:
            # Get the user's input - this returns a string with all text parts combined
            user_text = context.get_user_input()
            
            # Handle the case where no text was found
            if not user_text or user_text.strip() == "":
                # Create an error message using the SDK utility
                error_message = new_agent_text_message(
                    "I didn't receive any text to echo. Please send me a message with text content."
                )
                await event_queue.enqueue_event(error_message)
                return
            
            # Create and enqueue the echo response using the SDK utility
            echo_message = new_agent_text_message(f"You said: '{user_text.strip()}'")
            await event_queue.enqueue_event(echo_message)
            
        except Exception as e:
            # The SDK provides structured error handling
            error_message = new_agent_text_message(f"Error processing your message: {str(e)}")
            await event_queue.enqueue_event(error_message)
    
    async def cancel(self, task_id, event_queue):
        """
        Handle task cancellation using the SDK interface.
        
        Args:
            task_id: The ID of the task being canceled.
            event_queue: EventQueue for publishing cancellation events.
        """
        cancel_message = new_agent_text_message(f"Echo task {task_id} has been canceled.")
        await event_queue.enqueue_event(cancel_message)

def create_app():
    """Create and configure the A2A application using real SDK components."""
    
    # Create the agent executor
    executor = EchoAgentExecutor()
    
    # Create task store for managing task state
    task_store = InMemoryTaskStore()
    
    # Create the request handler that coordinates everything
    request_handler = DefaultRequestHandler(
        agent_executor=executor,
        task_store=task_store
    )
    
    # Create the A2A FastAPI application
    app = A2AFastAPIApplication(
        agent_card=agent_card,
        http_handler=request_handler
    )
    
    # Build and return the configured FastAPI app
    return app.build()

if __name__ == "__main__":
    print("🚀 Starting A2A SDK Echo Agent...")
    print("📡 Agent Card: http://localhost:8000/.well-known/agent-card.json")
    print("🔗 A2A Endpoint: http://localhost:8000/")
    
    app = create_app()
    
    uvicorn.run(
        app,
        host="0.0.0.0",
        port=8000,
        log_level="info"
    )
Testing with previously created client

We can see that if we try to use the original previous client (simple_a2a_client.py) with the SDK server, we’ll get an error because it expects a Task response with status and history fields. The SDK returns direct Message responses instead of Task-wrapped responses like the raw implementation. This makes the SDK more efficient, but requires clients to handle the improved response format. Both implementations follow the A2A protocol for discovery and messaging, but they differ in their response structures.

The SDK’s direct message format is more efficient.

  • Raw implementation: {"result": {"status": {...}, "history": [...]}}.

  • SDK implementation: {"result": {"kind": "message", "parts": [...], "role": "agent"}}.

This demonstrates the SDK's improved architecture, with less wrapper overhead, and more direct communication. For completeness, here’s the SDK-compatible client (simple_a2a_client_sdk.py) that handles both response formats:

import requests
import uuid
import json

def main():
    # Step 1: Discover the Agent
    base_url = "http://localhost:8000"
    print("🔍 Discovering A2A agent...")
    
    try:
        agent_card = requests.get(f"{base_url}/.well-known/agent-card.json").json()
        print(f"✅ Found: {agent_card['name']} - {agent_card['description']}")
    except requests.RequestException as e:
        print(f"❌ Discovery failed: {e}")
        return
    
    # Step 2: Send a Message using A2A JSON-RPC
    print("\n💬 Sending message...")
    
    message = {
        "jsonrpc": "2.0",
        "id": str(uuid.uuid4()),
        "method": "message/send",
        "params": {
            "message": {
                "kind": "message",
                "messageId": str(uuid.uuid4()),
                "role": "user",
                "parts": [{"kind": "text", "text": "Hello A2A world!"}]
            }
        }
    }
    
    try:
        response = requests.post(base_url, json=message).json()
    except requests.RequestException as e:
        print(f"❌ Communication failed: {e}")
        return
    
    # Step 3: Handle the Response
    # The SDK returns Message directly, not Task with history like raw implementation
    if "result" in response:
        result = response["result"] 
        if isinstance(result, dict) and "parts" in result:
            # This is a direct Message response from SDK
            agent_reply = result["parts"][0]["text"]
            print(f"🤖 Agent: {agent_reply}")
        else:
            # This might be a Task response (for compatibility)
            print(f"📋 Task Status: {result.get('status', {}).get('state', 'completed')}")
            
            # Find agent's reply in history
            for msg in result.get("history", []):
                if msg.get("role") == "agent":
                    agent_reply = msg["parts"][0]["text"]
                    print(f"🤖 Agent: {agent_reply}")
                    break
    else:
        error = response.get("error", {})
        print(f"❌ Error: {error.get('message', 'Unknown error')}")
    
    print("\n✨ A2A communication complete!")

if __name__ == "__main__":
    main()
Testing with A2A SDK-compliant client

This client gracefully handles both response formats, making it compatible with both the raw implementation and the SDK implementation.

What are the differences between the two implementations?

While the SDK implementation might seem more complex at first, with its multiple classes and event-driven architecture, it is designed for simplicity at scale. The raw FastAPI approach forces you to handle every protocol detail manually, which becomes increasingly burdensome as your agent’s functionality grows. The SDK abstracts away this complexity, letting you focus on your agent’s unique capabilities while providing a robust foundation that scales smoothly from simple echo agents to sophisticated multi-agent systems.

Aspect

Raw Implementation

SDK Implementation

JSON-RPC handling

Manual parsing and formatting

Automatic

Message validation

Custom Pydantic models

Built-in

Task life cycle

Manual state management

Automatic

Error handling

Custom JSON-RPC errors

SDK utility functions

Agent discovery

Manual endpoint setup

Automatic

Response construction

Manual JSON building

Event queue abstraction

Type safety

Limited validation

Full SDK type checking

Testing

Mock HTTP requests

SDK test utilities

Observability

Custom logging

Built-in tracing hooks

Authentication

Not implemented

SDK framework support

The SDK’s true power becomes clear when you need to scale beyond a simple echo. Imagine building a travel-planning agent that coordinates with flight-booking services, hotel reservations, and local activity providers. With the raw implementation, you would have to manually handle streaming responses for real-time price updates, and manage complex multi-turn conversations across different services. You would also have to implement authentication for each provider, and build custom error recovery for network failures. The SDK manages all of this automatically: streaming through the event queue, conversation state through the task store, authentication through pluggable handlers, and resilient error handling through structured exceptions.

This architectural foundation means that adding new capabilities such as file processing, multilingual support, or integration with external APIs becomes a matter of extending your AgentExecutor rather than rewriting protocol infrastructure. The same patterns that power a simple echo agent scale seamlessly to enterprise-grade systems, handling thousands of concurrent conversations across multiple channels.