A2A Protocol Documentation

Appearance

A2A Protocol Specification (Python) ​

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 ​

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

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

Role (Enum) ​

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

TaskState (Enum) ​

python
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

AgentCard ​

python
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 ​

python
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 ​

python
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 ​

python
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 ​

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

AgentInterface ​

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

Message ​

python
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) ​

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

TextPart ​

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

FilePart ​

python
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 ​

python
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 ​
python
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 ​
python
class FileWithUri(A2ABaseModel):
    """File containing URI"""
    uri: str                          # File URL
    name: str | None = None           # File name
    mimeType: str | None = None       # MIME type

Task ​

python
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 ​

python
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 ​

python
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

JSONRPCMessage (Base Class) ​

python
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 ​

python
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 ​

python
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 ​

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

A2A Specific Errors ​

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

SecurityScheme (Union Type) ​

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

APIKeySecurityScheme ​

python
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 ​

python
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 ​

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

OpenIdConnectSecurityScheme ​

python
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 ​
python
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

PushNotificationConfig ​

python
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 ​

python
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 ​
python
class SendMessageRequest(A2ABaseModel):
    """Send message request"""
    jsonrpc: Literal['2.0'] = '2.0'
    method: Literal['message/send'] = 'message/send'
    params: MessageSendParams
    id: str | int
MessageSendParams ​
python
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 ​
python
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 ​
python
class GetTaskRequest(A2ABaseModel):
    """Get task request"""
    jsonrpc: Literal['2.0'] = '2.0'
    method: Literal['tasks/get'] = 'tasks/get'
    params: TaskQueryParams
    id: str | int
TaskQueryParams ​
python
class TaskQueryParams(A2ABaseModel):
    """Task query parameters"""
    id: str                              # Task ID
    historyLength: int | None = None     # History length
    metadata: dict[str, Any] | None = None  # Metadata
CancelTaskRequest ​
python
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 ​

python
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 ​

python
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 ​

Protocol Flow Diagram ​

Task State Transition Diagram ​

Message Part Type Hierarchy Diagram ​

Security Scheme Type Diagram ​

Error Code Mapping Table ​

Error CodeError TypeDescriptionUse Case
-32700JSONParseErrorJSON parse errorInvalid JSON format
-32600InvalidRequestErrorInvalid requestRequest format doesn't comply with specification
-32601MethodNotFoundErrorMethod not foundCalling non-existent method
-32602InvalidParamsErrorInvalid parametersIncorrect method parameters
-32603InternalErrorInternal errorServer internal processing error
-32001TaskNotFoundErrorTask not foundRequested task ID doesn't exist
-32002TaskNotCancelableErrorTask not cancelableTask state doesn't allow cancellation
-32003PushNotificationNotSupportedErrorPush notification not supportedAgent doesn't support push notifications
-32004UnsupportedOperationErrorOperation not supportedAgent doesn't support requested operation
-32005ContentTypeNotSupportedErrorContent type not supportedRequested content type doesn't match agent capabilities
-32006InvalidAgentResponseErrorInvalid agent responseAgent returned incorrectly formatted response

Supported Methods List ​

The A2A protocol defines the following standard methods:

  • message/send - Send message to agent
  • message/stream - Send streaming message to agent
  • 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 ​

python
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 ​

python
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 ​

python
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 ​

python
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

A2A Protocol Documentation

Copyright © 2025 a2aprotocol.ai