██FR█████ █INTELL███████████

A2A Protocol Specification (Python)


This content originally appeared on DEV Community and was authored by cz

Overview

The A2A (Agent2Agent) protocol is a JSON-RPC 2.0 based communication protocol designed for interactions between intelligent agents. This protocol defines core functionalities including agent capability declarations, message passing, task management, and security authentication.

This document is based on the a2a-python/src/a2a/types.py file and provides a detailed introduction to all data structures and object relationships in the A2A protocol.

Protocol Version

Current protocol version: 0.2.5

Core Concepts

1. Agent

An agent is the core entity in the A2A protocol, possessing specific skills and capabilities to handle user requests and execute tasks.

2. Task

A task is a unit of work executed by an agent, featuring lifecycle state management and historical record functionality.

3. Message

A message is the basic unit for exchanging information between users and agents, supporting multiple content types.

4. JSON-RPC 2.0

The A2A protocol is based on the JSON-RPC 2.0 standard, providing standardized request-response communication patterns.

Data Structure Details

Basic Types

A2A 

class A2A(RootModel[Any]):
    root: Any

The root type of the protocol, can contain arbitrary data.

Role (Enum) 

class Role(str, Enum):
    """Role of the message sender"""
    agent = 'agent'  # Agent
    user = 'user'    # User

TaskState (Enum) 

class TaskState(str, Enum):
    """Possible states of a task"""
    submitted = 'submitted'           # Submitted
    working = 'working'               # Working
    input_required = 'input-required' # Input required
    completed = 'completed'           # Completed
    canceled = 'canceled'             # Canceled
    failed = 'failed'                 # Failed
    rejected = 'rejected'             # Rejected
    auth_required = 'auth-required'   # Authentication required
    unknown = 'unknown'               # Unknown state

Agent-Related Types

AgentCard 

class AgentCard(A2ABaseModel):
    """Agent card - contains key information about the agent"""
    name: str                                    # Agent name
    description: str                             # Agent description
    version: str                                 # Agent version
    url: str                                     # Agent URL
    protocolVersion: str | None = '0.2.5'       # Protocol version
    skills: list[AgentSkill]                     # Skills list
    capabilities: AgentCapabilities              # Capability declarations
    defaultInputModes: list[str]                 # Default input modes
    defaultOutputModes: list[str]                # Default output modes
    provider: AgentProvider | None = None        # Service provider
    security: list[dict[str, list[str]]] | None = None  # Security requirements
    securitySchemes: dict[str, SecurityScheme] | None = None  # Security schemes
    # ... other fields

AgentSkill 

class AgentSkill(A2ABaseModel):
    """Agent skill - capability units that the agent can execute"""
    id: str                          # Skill unique identifier
    name: str                        # Skill name
    description: str                 # Skill description
    tags: list[str]                  # Skill tags
    examples: list[str] | None = None # Usage examples
    inputModes: list[str] | None = None   # Input modes
    outputModes: list[str] | None = None  # Output modes

AgentCapabilities 

class AgentCapabilities(A2ABaseModel):
    """Agent capability definitions"""
    extensions: list[AgentExtension] | None = None      # Supported extensions
    pushNotifications: bool | None = None               # Push notification support
    stateTransitionHistory: bool | None = None          # State transition history
    streaming: bool | None = None                       # Streaming support

AgentExtension 

class AgentExtension(A2ABaseModel):
    """Agent extension declaration"""
    uri: str                                # Extension URI
    description: str | None = None          # Extension description
    params: dict[str, Any] | None = None    # Extension parameters
    required: bool | None = None            # Whether required

AgentProvider 

class AgentProvider(A2ABaseModel):
    """Agent service provider"""
    organization: str  # Organization name
    url: str          # Organization URL

AgentInterface 

class AgentInterface(A2ABaseModel):
    """Agent interface declaration"""
    transport: str  # Transport protocol (JSONRPC, GRPC, HTTP+JSON)
    url: str       # Interface URL

Message-Related Types

Message 

class Message(A2ABaseModel):
    """Messages exchanged between users and agents"""
    messageId: str                           # Message ID
    role: Role                               # Sender role
    parts: list[Part]                        # Message content parts
    kind: Literal['message'] = 'message'     # Event type
    taskId: str | None = None                # Associated task ID
    contextId: str | None = None             # Context ID
    referenceTaskIds: list[str] | None = None # Referenced task ID list
    extensions: list[str] | None = None       # Extension URI list
    metadata: dict[str, Any] | None = None    # Metadata

Part (Union Type) 

class Part(RootModel[TextPart | FilePart | DataPart]):
    """Message part - can be text, file, or structured data"""
    root: TextPart | FilePart | DataPart

TextPart 

class TextPart(A2ABaseModel):
    """Text message part"""
    kind: Literal['text'] = 'text'           # Part type
    text: str                                # Text content
    metadata: dict[str, Any] | None = None   # Metadata

FilePart 

class FilePart(A2ABaseModel):
    """File message part"""
    kind: Literal['file'] = 'file'           # Part type
    file: FileWithBytes | FileWithUri        # File content
    metadata: dict[str, Any] | None = None   # Metadata

DataPart 

class DataPart(A2ABaseModel):
    """Structured data message part"""
    kind: Literal['data'] = 'data'           # Part type
    data: dict[str, Any]                     # Structured data
    metadata: dict[str, Any] | None = None   # Metadata

File Types

FileWithBytes 
class FileWithBytes(A2ABaseModel):
    """File containing byte content"""
    bytes: str                        # base64-encoded file content
    name: str | None = None           # File name
    mimeType: str | None = None       # MIME type
FileWithUri 
class FileWithUri(A2ABaseModel):
    """File containing URI"""
    uri: str                          # File URL
    name: str | None = None           # File name
    mimeType: str | None = None       # MIME type

Task-Related Types

Task 

class Task(A2ABaseModel):
    """Task entity"""
    id: str                                  # Task unique identifier
    contextId: str                           # Context ID
    status: TaskStatus                       # Task status
    kind: Literal['task'] = 'task'           # Event type
    history: list[Message] | None = None     # Message history
    artifacts: list[Artifact] | None = None  # Generated artifacts
    metadata: dict[str, Any] | None = None   # Metadata

TaskStatus 

class TaskStatus(A2ABaseModel):
    """Task status and related messages"""
    state: TaskState                         # Task state
    message: Message | None = None           # Status update message
    timestamp: str | None = None             # Timestamp (ISO 8601)

Artifact 

class Artifact(A2ABaseModel):
    """Artifacts generated by tasks"""
    artifactId: str                          # Artifact ID
    parts: list[Part]                        # Artifact content parts
    name: str | None = None                  # Artifact name
    description: str | None = None           # Artifact description
    extensions: list[str] | None = None      # Extension URI list
    metadata: dict[str, Any] | None = None   # Metadata

JSON-RPC Related Types

JSONRPCMessage (Base Class) 

class JSONRPCMessage(A2ABaseModel):
    """JSON-RPC 2.0 message base class"""
    jsonrpc: Literal['2.0'] = '2.0'          # Protocol version
    id: str | int | None = None              # Request/response ID

JSONRPCRequest 

class JSONRPCRequest(A2ABaseModel):
    """JSON-RPC 2.0 request object"""
    jsonrpc: Literal['2.0'] = '2.0'          # Protocol version
    method: str                              # Method name
    params: dict[str, Any] | None = None     # Parameters
    id: str | int | None = None              # Request ID

JSONRPCSuccessResponse 

class JSONRPCSuccessResponse(A2ABaseModel):
    """JSON-RPC 2.0 success response object"""
    jsonrpc: Literal['2.0'] = '2.0'          # Protocol version
    result: Any                              # Result data
    id: str | int | None = None              # Response ID

JSONRPCErrorResponse 

class JSONRPCErrorResponse(A2ABaseModel):
    """JSON-RPC 2.0 error response object"""
    jsonrpc: Literal['2.0'] = '2.0'          # Protocol version
    error: JSONRPCError | [specific error type]      # Error information
    id: str | int | None = None              # Response ID

Error Types

The A2A protocol defines multiple error types, all inheriting from standard JSON-RPC errors:

Standard JSON-RPC Errors

JSONParseError 
class JSONParseError(A2ABaseModel):
    """JSON parse error"""
    code: Literal[-32700] = -32700
    message: str | None = 'Invalid JSON payload'
    data: Any | None = None
InvalidRequestError 
class InvalidRequestError(A2ABaseModel):
    """Invalid request error"""
    code: Literal[-32600] = -32600
    message: str | None = 'Request payload validation error'
    data: Any | None = None
MethodNotFoundError 
class MethodNotFoundError(A2ABaseModel):
    """Method not found error"""
    code: Literal[-32601] = -32601
    message: str | None = 'Method not found'
    data: Any | None = None
InvalidParamsError 
class InvalidParamsError(A2ABaseModel):
    """Invalid parameters error"""
    code: Literal[-32602] = -32602
    message: str | None = 'Invalid parameters'
    data: Any | None = None
InternalError 
class InternalError(A2ABaseModel):
    """Internal error"""
    code: Literal[-32603] = -32603
    message: str | None = 'Internal error'
    data: Any | None = None

A2A Specific Errors

TaskNotFoundError 
class TaskNotFoundError(A2ABaseModel):
    """Task not found error"""
    code: Literal[-32001] = -32001
    message: str | None = 'Task not found'
    data: Any | None = None
TaskNotCancelableError 
class TaskNotCancelableError(A2ABaseModel):
    """Task not cancelable error"""
    code: Literal[-32002] = -32002
    message: str | None = 'Task cannot be canceled'
    data: Any | None = None
PushNotificationNotSupportedError 
class PushNotificationNotSupportedError(A2ABaseModel):
    """Push notification not supported error"""
    code: Literal[-32003] = -32003
    message: str | None = 'Push Notification is not supported'
    data: Any | None = None
UnsupportedOperationError 
class UnsupportedOperationError(A2ABaseModel):
    """Unsupported operation error"""
    code: Literal[-32004] = -32004
    message: str | None = 'This operation is not supported'
    data: Any | None = None
ContentTypeNotSupportedError 
class ContentTypeNotSupportedError(A2ABaseModel):
    """Content type not supported error"""
    code: Literal[-32005] = -32005
    message: str | None = 'Incompatible content types'
    data: Any | None = None
InvalidAgentResponseError 
class InvalidAgentResponseError(A2ABaseModel):
    """Invalid agent response error"""
    code: Literal[-32006] = -32006
    message: str | None = 'Invalid agent response'
    data: Any | None = None

Security Authentication Related Types

SecurityScheme (Union Type) 

class SecurityScheme(RootModel[
    APIKeySecurityScheme |
    HTTPAuthSecurityScheme |
    OAuth2SecurityScheme |
    OpenIdConnectSecurityScheme
]):
    """Security scheme - supports multiple authentication methods"""

APIKeySecurityScheme 

class APIKeySecurityScheme(A2ABaseModel):
    """API key security scheme"""
    type: Literal['apiKey'] = 'apiKey'
    name: str                              # Parameter name
    in_: In                                # Location (header/query/cookie)
    description: str | None = None         # Description

HTTPAuthSecurityScheme 

class HTTPAuthSecurityScheme(A2ABaseModel):
    """HTTP authentication security scheme"""
    type: Literal['http'] = 'http'
    scheme: str                            # Authentication scheme (Basic, Bearer, etc.)
    bearerFormat: str | None = None        # Bearer token format
    description: str | None = None         # Description

OAuth2SecurityScheme 

class OAuth2SecurityScheme(A2ABaseModel):
    """OAuth2.0 security scheme"""
    type: Literal['oauth2'] = 'oauth2'
    flows: OAuthFlows                      # OAuth flow configuration
    description: str | None = None         # Description

OpenIdConnectSecurityScheme 

class OpenIdConnectSecurityScheme(A2ABaseModel):
    """OpenID Connect security scheme"""
    type: Literal['openIdConnect'] = 'openIdConnect'
    openIdConnectUrl: str                  # OpenID Connect discovery URL
    description: str | None = None         # Description

OAuth Flow Types

OAuthFlows 
class OAuthFlows(A2ABaseModel):
    """OAuth flow configuration"""
    implicit: ImplicitOAuthFlow | None = None                    # Implicit flow
    password: PasswordOAuthFlow | None = None                    # Password flow
    clientCredentials: ClientCredentialsOAuthFlow | None = None  # Client credentials flow
    authorizationCode: AuthorizationCodeOAuthFlow | None = None  # Authorization code flow

Push Notification Related Types

PushNotificationConfig 

class PushNotificationConfig(A2ABaseModel):
    """Push notification configuration"""
    url: str                                                    # Push URL
    id: str | None = None                                       # Push notification ID
    token: str | None = None                                    # Session token
    authentication: PushNotificationAuthenticationInfo | None = None  # Authentication info

PushNotificationAuthenticationInfo 

class PushNotificationAuthenticationInfo(A2ABaseModel):
    """Push notification authentication information"""
    schemes: list[str]                     # Supported authentication schemes
    credentials: str | None = None         # Credentials

Request and Response Types

Message Sending

SendMessageRequest 
class SendMessageRequest(A2ABaseModel):
    """Send message request"""
    jsonrpc: Literal['2.0'] = '2.0'
    method: Literal['message/send'] = 'message/send'
    params: MessageSendParams
    id: str | int
MessageSendParams 
class MessageSendParams(A2ABaseModel):
    """Send message parameters"""
    message: Message                                    # Message to send
    configuration: MessageSendConfiguration | None = None  # Send configuration
    metadata: dict[str, Any] | None = None             # Metadata
MessageSendConfiguration 
class MessageSendConfiguration(A2ABaseModel):
    """Message send configuration"""
    acceptedOutputModes: list[str]                          # Accepted output modes
    blocking: bool | None = None                            # Whether to block request
    historyLength: int | None = None                        # History message length
    pushNotificationConfig: PushNotificationConfig | None = None  # Push notification config

Task Operations

GetTaskRequest 
class GetTaskRequest(A2ABaseModel):
    """Get task request"""
    jsonrpc: Literal['2.0'] = '2.0'
    method: Literal['tasks/get'] = 'tasks/get'
    params: TaskQueryParams
    id: str | int
TaskQueryParams 
class TaskQueryParams(A2ABaseModel):
    """Task query parameters"""
    id: str                              # Task ID
    historyLength: int | None = None     # History length
    metadata: dict[str, Any] | None = None  # Metadata
CancelTaskRequest 
class CancelTaskRequest(A2ABaseModel):
    """Cancel task request"""
    jsonrpc: Literal['2.0'] = '2.0'
    method: Literal['tasks/cancel'] = 'tasks/cancel'
    params: TaskIdParams
    id: str | int

Event Types

TaskStatusUpdateEvent 

class TaskStatusUpdateEvent(A2ABaseModel):
    """Task status update event"""
    kind: Literal['status-update'] = 'status-update'
    taskId: str                              # Task ID
    contextId: str                           # Context ID
    status: TaskStatus                       # Task status
    final: bool                              # Whether this is the final event
    metadata: dict[str, Any] | None = None   # Metadata

TaskArtifactUpdateEvent 

class TaskArtifactUpdateEvent(A2ABaseModel):
    """Task artifact update event"""
    kind: Literal['artifact-update'] = 'artifact-update'
    taskId: str                              # Task ID
    contextId: str                           # Context ID
    artifact: Artifact                       # Artifact
    append: bool | None = None               # Whether to append
    lastChunk: bool | None = None            # Whether this is the last chunk
    metadata: dict[str, Any] | None = None   # Metadata

Object Relationship Diagram 

graph TB
    %% Core entities
    AgentCard[AgentCard<br/>Agent Card]
    Task[Task<br/>Task]
    Message[Message<br/>Message]

    %% Agent-related
    AgentSkill[AgentSkill<br/>Agent Skill]
    AgentCapabilities[AgentCapabilities<br/>Agent Capabilities]
    AgentProvider[AgentProvider<br/>Service Provider]
    AgentExtension[AgentExtension<br/>Agent Extension]
    AgentInterface[AgentInterface<br/>Agent Interface]

    %% Message content
    Part[Part<br/>Message Part]
    TextPart[TextPart<br/>Text Part]
    FilePart[FilePart<br/>File Part]
    DataPart[DataPart<br/>Data Part]

    %% File types
    FileWithBytes[FileWithBytes<br/>Byte File]
    FileWithUri[FileWithUri<br/>URI File]

    %% Task-related
    TaskStatus[TaskStatus<br/>Task Status]
    Artifact[Artifact<br/>Artifact]

    %% JSON-RPC
    JSONRPCRequest[JSONRPCRequest<br/>JSON-RPC Request]
    JSONRPCResponse[JSONRPCResponse<br/>JSON-RPC Response]

    %% Security authentication
    SecurityScheme[SecurityScheme<br/>Security Scheme]
    APIKeySecurityScheme[APIKeySecurityScheme<br/>API Key Scheme]
    HTTPAuthSecurityScheme[HTTPAuthSecurityScheme<br/>HTTP Auth Scheme]
    OAuth2SecurityScheme[OAuth2SecurityScheme<br/>OAuth2 Scheme]
    OpenIdConnectSecurityScheme[OpenIdConnectSecurityScheme<br/>OpenID Connect Scheme]

    %% Push notifications
    PushNotificationConfig[PushNotificationConfig<br/>Push Notification Config]
    PushNotificationAuthenticationInfo[PushNotificationAuthenticationInfo<br/>Push Auth Info]

    %% Error types
    A2AError[A2AError<br/>A2A Error]
    JSONRPCError[JSONRPCError<br/>JSON-RPC Error]

    %% Events
    TaskStatusUpdateEvent[TaskStatusUpdateEvent<br/>Status Update Event]
    TaskArtifactUpdateEvent[TaskArtifactUpdateEvent<br/>Artifact Update Event]

    %% Relationship connections
    AgentCard --> AgentSkill
    AgentCard --> AgentCapabilities
    AgentCard --> AgentProvider
    AgentCard --> AgentInterface
    AgentCard --> SecurityScheme

    AgentCapabilities --> AgentExtension

    Task --> TaskStatus
    Task --> Message
    Task --> Artifact

    Message --> Part
    Part --> TextPart
    Part --> FilePart
    Part --> DataPart

    FilePart --> FileWithBytes
    FilePart --> FileWithUri

    Artifact --> Part

    SecurityScheme --> APIKeySecurityScheme
    SecurityScheme --> HTTPAuthSecurityScheme
    SecurityScheme --> OAuth2SecurityScheme
    SecurityScheme --> OpenIdConnectSecurityScheme

    PushNotificationConfig --> PushNotificationAuthenticationInfo

    TaskStatusUpdateEvent --> TaskStatus
    TaskArtifactUpdateEvent --> Artifact

    JSONRPCResponse --> A2AError
    A2AError --> JSONRPCError

    %% Styles
    classDef coreEntity fill:#e1f5fe
    classDef agentRelated fill:#f3e5f5
    classDef messageRelated fill:#e8f5e8
    classDef taskRelated fill:#fff3e0
    classDef securityRelated fill:#fce4ec
    classDef errorRelated fill:#ffebee
    classDef eventRelated fill:#f1f8e9

    class AgentCard,Task,Message coreEntity
    class AgentSkill,AgentCapabilities,AgentProvider,AgentExtension,AgentInterface agentRelated
    class Part,TextPart,FilePart,DataPart,FileWithBytes,FileWithUri messageRelated
    class TaskStatus,Artifact taskRelated
    class SecurityScheme,APIKeySecurityScheme,HTTPAuthSecurityScheme,OAuth2SecurityScheme,OpenIdConnectSecurityScheme,PushNotificationConfig,PushNotificationAuthenticationInfo securityRelated
    class A2AError,JSONRPCError errorRelated
    class TaskStatusUpdateEvent,TaskArtifactUpdateEvent eventRelated

Protocol Flow Diagram 

sequenceDiagram
    participant Client as Client
    participant Agent as Agent

    Note over Client,Agent: 1. Agent Discovery and Capability Query
    Client->>Agent: GET /agent-card
    Agent->>Client: AgentCard (skills, capabilities, security requirements)

    Note over Client,Agent: 2. Authentication (if required)
    Client->>Agent: Authentication request (according to SecurityScheme)
    Agent->>Client: Authentication response

    Note over Client,Agent: 3. Send Message and Create Task
    Client->>Agent: SendMessageRequest
    Note right of Agent: Create Task<br/>State: submitted
    Agent->>Client: SendMessageResponse (Task)

    Note over Client,Agent: 4. Task Processing (optional push notifications)
    loop Task Processing
        Note right of Agent: Update task state<br/>working -> completed
        alt Push notifications supported
            Agent->>Client: TaskStatusUpdateEvent
        else Polling mode
            Client->>Agent: GetTaskRequest
            Agent->>Client: GetTaskResponse (updated Task)
        end
    end

    Note over Client,Agent: 5. Get Final Results
    Client->>Agent: GetTaskRequest
    Agent->>Client: Task (with Artifacts and complete history)

Task State Transition Diagram 

stateDiagram-v2
    [*] --> submitted: Create task

    submitted --> working: Start processing
    submitted --> rejected: Reject task
    submitted --> auth_required: Authentication required

    working --> completed: Processing complete
    working --> failed: Processing failed
    working --> input_required: User input required
    working --> canceled: User canceled

    input_required --> working: Received user input
    input_required --> canceled: User canceled

    auth_required --> working: Authentication successful
    auth_required --> rejected: Authentication failed

    completed --> [*]
    failed --> [*]
    canceled --> [*]
    rejected --> [*]

    unknown --> working: State recovered
    unknown --> failed: Cannot recover

Message Part Type Hierarchy Diagram 

flowchart TD
    Part["Part<br/>Message Part Base Class"]

    Part --> TextPart["TextPart<br/>Text Part<br/>kind: text"]
    Part --> FilePart["FilePart<br/>File Part<br/>kind: file"]
    Part --> DataPart["DataPart<br/>Data Part<br/>kind: data"]

    FilePart --> FileContent{"File Content"}
    FileContent --> FileWithBytes["FileWithBytes<br/>Contains base64 byte data"]
    FileContent --> FileWithUri["FileWithUri<br/>Contains file URI"]

    TextPart --> TextContent["text: str<br/>Text content"]
    DataPart --> DataContent["data: dict str Any<br/>Structured data"]

    FileWithBytes --> BytesContent["bytes: str<br/>base64 encoded content"]
    FileWithUri --> UriContent["uri: str<br/>File URL"]

    %% Common attributes
    Part --> Metadata["metadata: dict str Any<br/>Optional metadata"]
    FileWithBytes --> FileMetadata["name: str<br/>mimeType: str<br/>File metadata"]
    FileWithUri --> FileMetadata

    classDef baseClass fill:#e3f2fd
    classDef textClass fill:#e8f5e8
    classDef fileClass fill:#fff3e0
    classDef dataClass fill:#f3e5f5
    classDef metaClass fill:#f5f5f5

    class Part baseClass
    class TextPart,TextContent textClass
    class FilePart,FileWithBytes,FileWithUri,BytesContent,UriContent fileClass
    class DataPart,DataContent dataClass
    class Metadata,FileMetadata metaClass

Security Scheme Type Diagram 

flowchart TD
    SecurityScheme["SecurityScheme<br/>Security Scheme Union Type"]

    SecurityScheme --> APIKeySecurityScheme["APIKeySecurityScheme<br/>API Key Authentication<br/>type: apiKey"]
    SecurityScheme --> HTTPAuthSecurityScheme["HTTPAuthSecurityScheme<br/>HTTP Authentication<br/>type: http"]
    SecurityScheme --> OAuth2SecurityScheme["OAuth2SecurityScheme<br/>OAuth2 Authentication<br/>type: oauth2"]
    SecurityScheme --> OpenIdConnectSecurityScheme["OpenIdConnectSecurityScheme<br/>OpenID Connect<br/>type: openIdConnect"]

    APIKeySecurityScheme --> APIKeyDetails["name: str<br/>in: header query cookie"]
    HTTPAuthSecurityScheme --> HTTPDetails["scheme: str<br/>bearerFormat: str"]
    OAuth2SecurityScheme --> OAuthFlows["flows: OAuthFlows"]
    OpenIdConnectSecurityScheme --> OIDCDetails["openIdConnectUrl: str"]

    OAuthFlows --> ImplicitFlow["implicit: ImplicitOAuthFlow"]
    OAuthFlows --> PasswordFlow["password: PasswordOAuthFlow"]
    OAuthFlows --> ClientCredentialsFlow["clientCredentials: ClientCredentialsOAuthFlow"]
    OAuthFlows --> AuthorizationCodeFlow["authorizationCode: AuthorizationCodeOAuthFlow"]

    ImplicitFlow --> ImplicitDetails["authorizationUrl: str<br/>scopes: dict str str"]
    PasswordFlow --> PasswordDetails["tokenUrl: str<br/>scopes: dict str str"]
    ClientCredentialsFlow --> ClientDetails["tokenUrl: str<br/>scopes: dict str str"]
    AuthorizationCodeFlow --> AuthCodeDetails["authorizationUrl: str<br/>tokenUrl: str<br/>scopes: dict str str"]

    classDef baseClass fill:#e3f2fd
    classDef apiKeyClass fill:#e8f5e8
    classDef httpClass fill:#fff3e0
    classDef oauthClass fill:#f3e5f5
    classDef oidcClass fill:#fce4ec

    class SecurityScheme baseClass
    class APIKeySecurityScheme,APIKeyDetails apiKeyClass
    class HTTPAuthSecurityScheme,HTTPDetails httpClass
    class OAuth2SecurityScheme,OAuthFlows,ImplicitFlow,PasswordFlow,ClientCredentialsFlow,AuthorizationCodeFlow,ImplicitDetails,PasswordDetails,ClientDetails,AuthCodeDetails oauthClass
    class OpenIdConnectSecurityScheme,OIDCDetails oidcClass

Error Code Mapping Table

Error Code Error Type Description Use Case
-32700 JSONParseError JSON parse error Invalid JSON format
-32600 InvalidRequestError Invalid request Request format doesn’t comply with specification
-32601 MethodNotFoundError Method not found Calling non-existent method
-32602 InvalidParamsError Invalid parameters Incorrect method parameters
-32603 InternalError Internal error Server internal processing error
-32001 TaskNotFoundError Task not found Requested task ID doesn’t exist
-32002 TaskNotCancelableError Task not cancelable Task state doesn’t allow cancellation
-32003 PushNotificationNotSupportedError Push notification not supported Agent doesn’t support push notifications
-32004 UnsupportedOperationError Operation not supported Agent doesn’t support requested operation
-32005 ContentTypeNotSupportedError Content type not supported Requested content type doesn’t match agent capabilities
-32006 InvalidAgentResponseError Invalid agent response Agent returned incorrectly formatted response

Supported Methods List

The A2A protocol defines the following standard methods:

Message-Related Methods

  • message/send – Send message to agent
  • message/stream – Send streaming message to agent

Task-Related Methods

  • tasks/get – Get task details
  • tasks/cancel – Cancel task
  • tasks/resubscribe – Resubscribe to task updates

Push Notification Configuration Methods

  • tasks/pushNotificationConfig/set – Set push notification configuration
  • tasks/pushNotificationConfig/get – Get push notification configuration
  • tasks/pushNotificationConfig/list – List push notification configurations
  • tasks/pushNotificationConfig/delete – Delete push notification configuration

Usage Examples

1. Send Simple Text Message 

from a2a.types import SendMessageRequest, MessageSendParams, Message, TextPart, Part

# Create text message part
text_part = TextPart(text="Hello, how can you help me?")
part = Part(root=text_part)

# Create message
message = Message(
    messageId="msg-001",
    role="user",
    parts=[part]
)

# Create send parameters
params = MessageSendParams(message=message)

# Create request
request = SendMessageRequest(
    id="req-001",
    params=params
)

2. Send Message with File 

from a2a.types import FilePart, FileWithBytes
import base64

# Read file and encode
with open("document.pdf", "rb") as f:
    file_bytes = base64.b64encode(f.read()).decode()

# Create file part
file_with_bytes = FileWithBytes(
    bytes=file_bytes,
    name="document.pdf",
    mimeType="application/pdf"
)

file_part = FilePart(file=file_with_bytes)
part = Part(root=file_part)

# Create message with file
message = Message(
    messageId="msg-002",
    role="user",
    parts=[part]
)

3. Query Task Status 

from a2a.types import GetTaskRequest, TaskQueryParams

# Create task query request
request = GetTaskRequest(
    id="req-002",
    params=TaskQueryParams(
        id="task-001",
        historyLength=10  # Get last 10 message history
    )
)

4. Configure Push Notifications 

from a2a.types import (
    SetTaskPushNotificationConfigRequest,
    TaskPushNotificationConfig,
    PushNotificationConfig,
    PushNotificationAuthenticationInfo
)

# Create push notification configuration
push_config = PushNotificationConfig(
    url="https://my-app.com/webhook/task-updates",
    authentication=PushNotificationAuthenticationInfo(
        schemes=["Bearer"],
        credentials="my-auth-token"
    )
)

# Create task push configuration
task_push_config = TaskPushNotificationConfig(
    taskId="task-001",
    pushNotificationConfig=push_config
)

# Create set request
request = SetTaskPushNotificationConfigRequest(
    id="req-003",
    params=task_push_config
)

Best Practices

1. Error Handling

  • Always check for error fields in responses
  • Implement appropriate retry strategies based on error codes
  • Provide meaningful error messages to users

2. Task Management

  • Use push notifications instead of polling for task updates
  • Implement task timeout mechanisms
  • Save task IDs for subsequent queries

3. Security

  • Use HTTPS transport
  • Properly implement authentication schemes
  • Regularly rotate API keys and tokens

4. Performance Optimization

  • Use streaming for handling large responses
  • Limit history message length
  • Implement client-side caching

Extensibility

The A2A protocol supports extensions through the following mechanisms:

  1. Agent Extensions – Declare custom capabilities through AgentExtension
  2. Metadata Fields – Most objects include optional metadata fields
  3. Custom Transport Protocols – Support new transport methods through AgentInterface
  4. Extension URIs – Reference extension specifications in messages and artifacts

Summary

The A2A protocol provides a comprehensive framework for communication and collaboration between intelligent agents. Through standardized message formats, task management, security authentication, and error handling, this protocol ensures interoperability between different agent implementations.

The protocol’s design considers extensibility and backward compatibility, enabling it to adapt to future requirement changes. By supporting multiple content types, transport protocols, and authentication schemes, the A2A protocol provides a solid foundation for building complex agent ecosystems.

A2A Specification


This content originally appeared on DEV Community and was authored by cz