DEV Community

Cover image for A2A Protocol Specification (Python)
cz
cz

Posted on

A2A Protocol Specification (Python)

#a2a #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 

class A2A(RootModel[Any]): root: Any
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

AgentProvider 

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

AgentInterface 

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

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
Enter fullscreen mode Exit fullscreen mode

Part (Union Type) 

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

TextPart 

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

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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

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
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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

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

Security Authentication Related Types

SecurityScheme (Union Type) 

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

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

OAuth2SecurityScheme 

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

OpenIdConnectSecurityScheme 

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

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

PushNotificationAuthenticationInfo 

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

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
Enter fullscreen mode Exit fullscreen mode
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
Enter fullscreen mode Exit fullscreen mode
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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode
TaskQueryParams 
class TaskQueryParams(A2ABaseModel): """Task query parameters""" id: str # Task ID  historyLength: int | None = None # History length  metadata: dict[str, Any] | None = None # Metadata
Enter fullscreen mode Exit fullscreen mode
CancelTaskRequest 
class CancelTaskRequest(A2ABaseModel): """Cancel task request""" jsonrpc: Literal['2.0'] = '2.0' method: Literal['tasks/cancel'] = 'tasks/cancel' params: TaskIdParams id: str | int
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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 )
Enter fullscreen mode Exit fullscreen mode

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] )
Enter fullscreen mode Exit fullscreen mode

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  ) )
Enter fullscreen mode Exit fullscreen mode

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 )
Enter fullscreen mode Exit fullscreen mode

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

Top comments (0)