A2A Python SDK

The AgentCard structure has been redesigned in v1.0 to support multiple transport bindings and improved capability definitions.

Major Changes:

• Transport Bindings: The url parameter is removed from AgentCard. Instead, use the supported_interfaces field, which contains a list of AgentInterface objects. Each AgentInterface defines a protocol_binding ('JSONRPC', 'HTTP+JSON', or 'GRPC') and its own url.
• Capabilities: AgentCapabilities.supports_authenticated_extended_card is renamed to AgentCapabilities.extended_agent_card. The input_modes and output_modes fields were removed from AgentCapabilities; use AgentCard.default_input_modes and AgentCard.default_output_modes for card-level defaults.
• Skill Examples: The examples parameter was removed from AgentCard and moved to AgentSkill.

from a2a.types import AgentCard, AgentCapabilities, AgentInterface, AgentSkill

skill = AgentSkill(
    id='hello_world',
    name='Hello World',
    description='Returns a Hello World message.',
    tags=['hello', 'world'],
    input_modes=['text/plain'],
    output_modes=['text/plain'],
    examples=['hello world', 'Hello, World!'],  # moved from AgentCard.examples
)

agent_card = AgentCard(
    name='Hello World Agent',
    supported_interfaces=[
        # JSON-RPC
        AgentInterface(
            protocol_binding='JSONRPC',
            url='http://localhost:41241/a2a/jsonrpc/',
        ),
        # GRPC
        AgentInterface(
            protocol_binding='GRPC',
            url='http://localhost:50051/a2a/grpc/',
        )
    ],
    version='0.0.1',
    default_input_modes=['text/plain'],
    default_output_modes=['text/plain'],
    capabilities=AgentCapabilities(
        streaming=True,
        extended_agent_card=True,
    ),
    skills=[skill],
)

To enable modern v1.0 clients and servers to interoperate with legacy v0.3 A2A systems, the SDK uses a phased transformation process. This process bridges the gap between three distinct data representations:

1. Legacy v0.3 Pydantic Models (types.py): Acts as the 'pivot' format. Legacy JSON-RPC and REST implementations use these models. They serve as the intermediary between old wire formats and the modern SDK.
2. Legacy v0.3 Protobuf Bindings (a2a_v0_3_pb2.py): Native Protobuf bindings for the legacy v0.3 gRPC protocol, used to decode incoming bytes from legacy gRPC clients or encode outbound bytes to legacy gRPC servers.
3. Current v1.0 Protobuf Bindings (a2a.types.a2a_pb2): The central source of truth for the modern SDK. All legacy payloads must be translated into these core objects to be processed by the modern AgentExecutor.

The transformation flow follows this path: Legacy gRPC Bytes $\rightarrow$ Legacy Pydantic Model $\rightarrow$ Modern v1.0 Protobuf.

In v1.0, the server strictly enforces A2A spec rules for SendStreamingMessage. AgentExecutor implementations must follow exactly one of two mutually exclusive streaming patterns. Mixing these patterns or violating the order of events will raise an InvalidAgentResponseError.

Supported Patterns

1. Message-only stream: Enqueue exactly one Message and then stop.
2. Task lifecycle stream: Enqueue a Task first, followed by zero or more TaskStatusUpdateEvent or TaskArtifactUpdateEvent objects until a terminal state is reached.

Common Violations (Raises InvalidAgentResponseError)

• Enqueueing a Message after a Task (mixing modes).
• Enqueueing more than one Message.
• Enqueueing a Task or update event after a Message.
• Enqueueing a TaskStatusUpdateEvent before the initial Task.

# Pattern A: Message-only stream — one Message, then done.
class GreetingExecutor(AgentExecutor):
    async def execute(self, context, event_queue):
        await event_queue.enqueue_event(
            new_text_message('Hello!', role=Role.ROLE_AGENT)
        )

# Pattern B: Task lifecycle stream — Task first, then updates.
class WorkflowExecutor(AgentExecutor):
    def __init__(self, agent):
        self._agent = agent

    async def execute(self, context, event_queue):
        task = context.current_task or new_task_from_user_message(context.message)
        await event_queue.enqueue_event(task)  # ✅ Task MUST be first

        await event_queue.enqueue_event(
            new_text_status_update_event(
                task_id=task.id,
                context_id=task.context_id,
                state=TaskState.TASK_STATE_WORKING,
                text='Processing...',
            )
        )

        result = await self._agent.invoke(context.message)
        await event_queue.enqueue_event(
            new_text_artifact_update_event(
                task_id=task.id,
                context_id=task.context_id,
                name='result',
                text=result,
            )
        )

        await event_queue.enqueue_event(
            new_text_status_update_event(
                task_id=task.id,
                context_id=task.context_id,
                state=TaskState.TASK_STATE_COMPLETED,
                text='Done!',
            )
        )

To migrate an Agent application from A2A protocol v0.3 to v1.0 without service interruption in a distributed environment, follow the 'Expand, Migrate, Contract' strategy.

Warning: Always back up your database before proceeding.

Step 1: Apply Schema Updates

Run the a2a-db tool to add new columns (owner, protocol_version, last_updated). These columns are nullable, so existing v0.3 code will continue to function.

uv run a2a-db --database-url "your-database-url"

Verify the schema version with:

uv run a2a-db current

Step 2: Rolling Deployment in Compatibility Mode

Deploy v1.0 SDK code but configure it to write in the legacy v0.3 format. This allows v0.3 instances still in the cluster to read data produced by new v1.0 instances.

To do this, pass the appropriate conversion functions to your Store constructors (see Compatibility Conversions for details).

Step 3: Transition to v1.0 Mode

Critical: Only perform this step once 100% of your instances are running v1.0 code. v0.3 code cannot read v1.0 native entries.

To transition, remove the core_to_model_conversion arguments from your Store constructors to revert to native v1.0 write behavior.

The transition from v0.3 to v1.0 introduces several breaking changes and architectural shifts:

• Protobuf Migration: Core types are now Protobuf-based instead of Pydantic. You cannot assign arbitrary attributes to these objects. Use google.protobuf.json_format.MessageToDict for dictionary conversion and .HasField('field_name') to check for optional fields.
• Enum Naming: All enum values have changed from snake_case to SCREAMING_SNAKE_CASE to comply with ProtoJSON.
• AgentCard Restructuring: The url field is replaced by supported_interfaces (a list of AgentInterface objects containing protocol_binding, protocol_version, and url). Input/output modes now live directly on AgentCard as default_input_modes and default_output_modes.
• AgentExecutor Streaming: Servers strictly enforce the A2A spec. An executor must enqueue either a single Message OR a Task followed by update events. Mixing these or sending updates before the initial Task results in InvalidAgentResponseError.
• Application Setup: Wrapper classes like A2AStarletteApplication have been removed. Use route factory functions (create_jsonrpc_routes(), create_rest_routes(), and create_agent_card_routes()) to compose routes into Starlette or FastAPI apps.
• Helper Consolidation: All utilities are now located in a2a.helpers.