概述

A2A（Agent2Agent）协议是一个基于JSON-RPC 2.0的通信协议，专为智能代理之间的交互而设计。该协议定义了包括代理能力声明、消息传递、任务管理和安全认证在内的核心功能。

本文档基于a2a-python/src/a2a/types.py文件，详细介绍了A2A协议中的所有数据结构和对象关系。

协议版本

当前协议版本：0.2.5

核心概念

1. 代理（Agent）

代理是A2A协议中的核心实体，具有特定的技能和能力来处理用户请求并执行任务。

2. 任务（Task）

任务是代理执行的工作单元，具有生命周期状态管理和历史记录功能。

3. 消息（Message）

消息是用户和代理之间交换信息的基本单位，支持多种内容类型。

4. JSON-RPC 2.0

A2A协议基于JSON-RPC 2.0标准，提供标准化的请求-响应通信模式。

数据结构详解

基础类型

A2A

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

协议的根类型，可以包含任意数据。

Role（枚举）

class Role(str, Enum):
    """消息发送者的角色"""
    agent = 'agent'  # 代理
    user = 'user'    # 用户

TaskState（枚举）

class TaskState(str, Enum):
    """任务的可能状态"""
    submitted = 'submitted'           # 已提交
    working = 'working'               # 工作中
    input_required = 'input-required' # 需要输入
    completed = 'completed'           # 已完成
    canceled = 'canceled'             # 已取消
    failed = 'failed'                 # 失败
    rejected = 'rejected'             # 被拒绝
    auth_required = 'auth-required'   # 需要认证
    unknown = 'unknown'               # 未知状态

代理相关类型

AgentCard

class AgentCard(A2ABaseModel):
    """代理卡片 - 包含代理的关键信息"""
    name: str                                    # 代理名称
    description: str                             # 代理描述
    version: str                                 # 代理版本
    url: str                                     # 代理URL
    protocolVersion: str | None = '0.2.5'       # 协议版本
    skills: list[AgentSkill]                     # 技能列表
    capabilities: AgentCapabilities              # 能力声明
    defaultInputModes: list[str]                 # 默认输入模式
    defaultOutputModes: list[str]                # 默认输出模式
    provider: AgentProvider | None = None        # 服务提供商
    security: list[dict[str, list[str]]] | None = None  # 安全要求
    securitySchemes: dict[str, SecurityScheme] | None = None  # 安全方案
    # ... 其他字段

AgentSkill

class AgentSkill(A2ABaseModel):
    """代理技能 - 代理可以执行的能力单元"""
    id: str                          # 技能唯一标识符
    name: str                        # 技能名称
    description: str                 # 技能描述
    tags: list[str]                  # 技能标签
    examples: list[str] | None = None # 使用示例
    inputModes: list[str] | None = None   # 输入模式
    outputModes: list[str] | None = None  # 输出模式

AgentCapabilities

class AgentCapabilities(A2ABaseModel):
    """代理能力定义"""
    extensions: list[AgentExtension] | None = None      # 支持的扩展
    pushNotifications: bool | None = None               # 推送通知支持
    stateTransitionHistory: bool | None = None          # 状态转换历史
    streaming: bool | None = None                       # 流式传输支持

AgentExtension

class AgentExtension(A2ABaseModel):
    """代理扩展声明"""
    uri: str                                # 扩展URI
    description: str | None = None          # 扩展描述
    params: dict[str, Any] | None = None    # 扩展参数
    required: bool | None = None            # 是否必需

AgentProvider

class AgentProvider(A2ABaseModel):
    """代理服务提供商"""
    organization: str  # 组织名称
    url: str          # 组织URL

AgentInterface

class AgentInterface(A2ABaseModel):
    """代理接口声明"""
    transport: str  # 传输协议（JSONRPC、GRPC、HTTP+JSON）
    url: str       # 接口URL

消息相关类型

Message

class Message(A2ABaseModel):
    """用户和代理之间交换的消息"""
    messageId: str                           # 消息ID
    role: Role                               # 发送者角色
    parts: list[Part]                        # 消息内容部分
    kind: Literal['message'] = 'message'     # 事件类型
    taskId: str | None = None                # 关联的任务ID
    contextId: str | None = None             # 上下文ID
    referenceTaskIds: list[str] | None = None # 引用的任务ID列表
    extensions: list[str] | None = None       # 扩展URI列表
    metadata: dict[str, Any] | None = None    # 元数据

Part（联合类型）

class Part(RootModel[TextPart | FilePart | DataPart]):
    """消息部分 - 可以是文本、文件或结构化数据"""
    root: TextPart | FilePart | DataPart

TextPart

class TextPart(A2ABaseModel):
    """文本消息部分"""
    kind: Literal['text'] = 'text'           # 部分类型
    text: str                                # 文本内容
    metadata: dict[str, Any] | None = None   # 元数据

FilePart

class FilePart(A2ABaseModel):
    """文件消息部分"""
    kind: Literal['file'] = 'file'           # 部分类型
    file: FileWithBytes | FileWithUri        # 文件内容
    metadata: dict[str, Any] | None = None   # 元数据

DataPart

class DataPart(A2ABaseModel):
    """结构化数据消息部分"""
    kind: Literal['data'] = 'data'           # 部分类型
    data: dict[str, Any]                     # 结构化数据
    metadata: dict[str, Any] | None = None   # 元数据

文件类型

FileWithBytes

class FileWithBytes(A2ABaseModel):
    """包含字节内容的文件"""
    bytes: str                        # base64编码的文件内容
    name: str | None = None           # 文件名
    mimeType: str | None = None       # MIME类型

FileWithUri

class FileWithUri(A2ABaseModel):
    """包含URI的文件"""
    uri: str                          # 文件URL
    name: str | None = None           # 文件名
    mimeType: str | None = None       # MIME类型

任务相关类型

Task

class Task(A2ABaseModel):
    """任务实体"""
    id: str                                  # 任务唯一标识符
    contextId: str                           # 上下文ID
    status: TaskStatus                       # 任务状态
    kind: Literal['task'] = 'task'           # 事件类型
    history: list[Message] | None = None     # 消息历史
    artifacts: list[Artifact] | None = None  # 生成的工件
    metadata: dict[str, Any] | None = None   # 元数据

TaskStatus

class TaskStatus(A2ABaseModel):
    """任务状态和相关消息"""
    state: TaskState                         # 任务状态
    message: Message | None = None           # 状态更新消息
    timestamp: str | None = None             # 时间戳（ISO 8601）

Artifact

class Artifact(A2ABaseModel):
    """任务生成的工件"""
    artifactId: str                          # 工件ID
    parts: list[Part]                        # 工件内容部分
    name: str | None = None                  # 工件名称
    description: str | None = None           # 工件描述
    extensions: list[str] | None = None      # 扩展URI列表
    metadata: dict[str, Any] | None = None   # 元数据

JSON-RPC相关类型

JSONRPCMessage（基类）

class JSONRPCMessage(A2ABaseModel):
    """JSON-RPC 2.0消息基类"""
    jsonrpc: Literal['2.0'] = '2.0'          # 协议版本
    id: str | int | None = None              # 请求/响应ID

JSONRPCRequest

class JSONRPCRequest(A2ABaseModel):
    """JSON-RPC 2.0请求对象"""
    jsonrpc: Literal['2.0'] = '2.0'          # 协议版本
    method: str                              # 方法名
    params: dict[str, Any] | None = None     # 参数
    id: str | int | None = None              # 请求ID

JSONRPCSuccessResponse

class JSONRPCSuccessResponse(A2ABaseModel):
    """JSON-RPC 2.0成功响应对象"""
    jsonrpc: Literal['2.0'] = '2.0'          # 协议版本
    result: Any                              # 结果数据
    id: str | int | None = None              # 响应ID

JSONRPCErrorResponse

class JSONRPCErrorResponse(A2ABaseModel):
    """JSON-RPC 2.0错误响应对象"""
    jsonrpc: Literal['2.0'] = '2.0'          # 协议版本
    error: JSONRPCError | [specific error type]      # 错误信息
    id: str | int | None = None              # 响应ID

错误类型

A2A协议定义了多种错误类型，都继承自标准JSON-RPC错误：

标准JSON-RPC错误

JSONParseError

class JSONParseError(A2ABaseModel):
    """JSON解析错误"""
    code: Literal[-32700] = -32700
    message: str | None = 'Invalid JSON payload'
    data: Any | None = None

InvalidRequestError

class InvalidRequestError(A2ABaseModel):
    """无效请求错误"""
    code: Literal[-32600] = -32600
    message: str | None = 'Request payload validation error'
    data: Any | None = None

MethodNotFoundError

class MethodNotFoundError(A2ABaseModel):
    """方法未找到错误"""
    code: Literal[-32601] = -32601
    message: str | None = 'Method not found'
    data: Any | None = None

InvalidParamsError

class InvalidParamsError(A2ABaseModel):
    """无效参数错误"""
    code: Literal[-32602] = -32602
    message: str | None = 'Invalid parameters'
    data: Any | None = None

InternalError

class InternalError(A2ABaseModel):
    """内部错误"""
    code: Literal[-32603] = -32603
    message: str | None = 'Internal error'
    data: Any | None = None

A2A特定错误

TaskNotFoundError

class TaskNotFoundError(A2ABaseModel):
    """任务未找到错误"""
    code: Literal[-32001] = -32001
    message: str | None = 'Task not found'
    data: Any | None = None

TaskNotCancelableError

class TaskNotCancelableError(A2ABaseModel):
    """任务不可取消错误"""
    code: Literal[-32002] = -32002
    message: str | None = 'Task cannot be canceled'
    data: Any | None = None

PushNotificationNotSupportedError

class PushNotificationNotSupportedError(A2ABaseModel):
    """推送通知不支持错误"""
    code: Literal[-32003] = -32003
    message: str | None = 'Push Notification is not supported'
    data: Any | None = None

UnsupportedOperationError

class UnsupportedOperationError(A2ABaseModel):
    """不支持的操作错误"""
    code: Literal[-32004] = -32004
    message: str | None = 'This operation is not supported'
    data: Any | None = None

ContentTypeNotSupportedError

class ContentTypeNotSupportedError(A2ABaseModel):
    """内容类型不支持错误"""
    code: Literal[-32005] = -32005
    message: str | None = 'Incompatible content types'
    data: Any | None = None

InvalidAgentResponseError

class InvalidAgentResponseError(A2ABaseModel):
    """无效代理响应错误"""
    code: Literal[-32006] = -32006
    message: str | None = 'Invalid agent response'
    data: Any | None = None

安全认证相关类型

SecurityScheme（联合类型）

class SecurityScheme(RootModel[
    APIKeySecurityScheme |
    HTTPAuthSecurityScheme |
    OAuth2SecurityScheme |
    OpenIdConnectSecurityScheme
]):
    """安全方案 - 支持多种认证方法"""

APIKeySecurityScheme

class APIKeySecurityScheme(A2ABaseModel):
    """API密钥安全方案"""
    type: Literal['apiKey'] = 'apiKey'
    name: str                              # 参数名称
    in_: In                                # 位置（header/query/cookie）
    description: str | None = None         # 描述

HTTPAuthSecurityScheme

class HTTPAuthSecurityScheme(A2ABaseModel):
    """HTTP认证安全方案"""
    type: Literal['http'] = 'http'
    scheme: str                            # 认证方案（Basic、Bearer等）
    bearerFormat: str | None = None        # Bearer令牌格式
    description: str | None = None         # 描述

OAuth2SecurityScheme

class OAuth2SecurityScheme(A2ABaseModel):
    """OAuth2.0安全方案"""
    type: Literal['oauth2'] = 'oauth2'
    flows: OAuthFlows                      # OAuth流程配置
    description: str | None = None         # 描述

OpenIdConnectSecurityScheme

class OpenIdConnectSecurityScheme(A2ABaseModel):
    """OpenID Connect安全方案"""
    type: Literal['openIdConnect'] = 'openIdConnect'
    openIdConnectUrl: str                  # OpenID Connect发现URL
    description: str | None = None         # 描述

OAuth流程类型

OAuthFlows

class OAuthFlows(A2ABaseModel):
    """OAuth流程配置"""
    implicit: ImplicitOAuthFlow | None = None                    # 隐式流程
    password: PasswordOAuthFlow | None = None                    # 密码流程
    clientCredentials: ClientCredentialsOAuthFlow | None = None  # 客户端凭据流程
    authorizationCode: AuthorizationCodeOAuthFlow | None = None  # 授权码流程

推送通知相关类型

PushNotificationConfig

class PushNotificationConfig(A2ABaseModel):
    """推送通知配置"""
    url: str                                                    # 推送URL
    id: str | None = None                                       # 推送通知ID
    token: str | None = None                                    # 会话令牌
    authentication: PushNotificationAuthenticationInfo | None = None  # 认证信息

PushNotificationAuthenticationInfo

class PushNotificationAuthenticationInfo(A2ABaseModel):
    """推送通知认证信息"""
    schemes: list[str]                     # 支持的认证方案
    credentials: str | None = None         # 凭据

请求和响应类型

消息发送

SendMessageRequest

class SendMessageRequest(A2ABaseModel):
    """发送消息请求"""
    jsonrpc: Literal['2.0'] = '2.0'
    method: Literal['message/send'] = 'message/send'
    params: MessageSendParams
    id: str | int

MessageSendParams

class MessageSendParams(A2ABaseModel):
    """发送消息参数"""
    message: Message                                    # 要发送的消息
    configuration: MessageSendConfiguration | None = None  # 发送配置
    metadata: dict[str, Any] | None = None             # 元数据

MessageSendConfiguration

class MessageSendConfiguration(A2ABaseModel):
    """消息发送配置"""
    acceptedOutputModes: list[str]                          # 接受的输出模式
    blocking: bool | None = None                            # 是否阻塞请求
    historyLength: int | None = None                        # 历史消息长度
    pushNotificationConfig: PushNotificationConfig | None = None  # 推送通知配置

任务操作

GetTaskRequest

class GetTaskRequest(A2ABaseModel):
    """获取任务请求"""
    jsonrpc: Literal['2.0'] = '2.0'
    method: Literal['tasks/get'] = 'tasks/get'
    params: TaskQueryParams
    id: str | int

TaskQueryParams

class TaskQueryParams(A2ABaseModel):
    """任务查询参数"""
    id: str                              # 任务ID
    historyLength: int | None = None     # 历史长度
    metadata: dict[str, Any] | None = None  # 元数据

CancelTaskRequest

class CancelTaskRequest(A2ABaseModel):
    """取消任务请求"""
    jsonrpc: Literal['2.0'] = '2.0'
    method: Literal['tasks/cancel'] = 'tasks/cancel'
    params: TaskIdParams
    id: str | int

事件类型

TaskStatusUpdateEvent

class TaskStatusUpdateEvent(A2ABaseModel):
    """任务状态更新事件"""
    kind: Literal['status-update'] = 'status-update'
    taskId: str                              # 任务ID
    contextId: str                           # 上下文ID
    status: TaskStatus                       # 任务状态
    final: bool                              # 是否为最终事件
    metadata: dict[str, Any] | None = None   # 元数据

TaskArtifactUpdateEvent

class TaskArtifactUpdateEvent(A2ABaseModel):
    """任务工件更新事件"""
    kind: Literal['artifact-update'] = 'artifact-update'
    taskId: str                              # 任务ID
    contextId: str                           # 上下文ID
    artifact: Artifact                       # 工件
    append: bool | None = None               # 是否追加
    lastChunk: bool | None = None            # 是否为最后一块
    metadata: dict[str, Any] | None = None   # 元数据

对象关系图

graph TB
    %% 核心实体
    AgentCard[AgentCard<br/>代理卡片]
    Task[Task<br/>任务]
    Message[Message<br/>消息]
    
    %% 代理相关
    AgentSkill[AgentSkill<br/>代理技能]
    AgentCapabilities[AgentCapabilities<br/>代理能力]
    AgentProvider[AgentProvider<br/>服务提供商]
    AgentExtension[AgentExtension<br/>代理扩展]
    AgentInterface[AgentInterface<br/>代理接口]
    
    %% 消息内容
    Part[Part<br/>消息部分]
    TextPart[TextPart<br/>文本部分]
    FilePart[FilePart<br/>文件部分]
    DataPart[DataPart<br/>数据部分]
    
    %% 文件类型
    FileWithBytes[FileWithBytes<br/>字节文件]
    FileWithUri[FileWithUri<br/>URI文件]
    
    %% 任务相关
    TaskStatus[TaskStatus<br/>任务状态]
    Artifact[Artifact<br/>工件]
    
    %% JSON-RPC
    JSONRPCRequest[JSONRPCRequest<br/>JSON-RPC请求]
    JSONRPCResponse[JSONRPCResponse<br/>JSON-RPC响应]
    
    %% 安全认证
    SecurityScheme[SecurityScheme<br/>安全方案]
    APIKeySecurityScheme[APIKeySecurityScheme<br/>API密钥方案]
    HTTPAuthSecurityScheme[HTTPAuthSecurityScheme<br/>HTTP认证方案]
    OAuth2SecurityScheme[OAuth2SecurityScheme<br/>OAuth2方案]
    OpenIdConnectSecurityScheme[OpenIdConnectSecurityScheme<br/>OpenID Connect方案]
    
    %% 推送通知
    PushNotificationConfig[PushNotificationConfig<br/>推送通知配置]
    PushNotificationAuthenticationInfo[PushNotificationAuthenticationInfo<br/>推送认证信息]
    
    %% 错误类型
    A2AError[A2AError<br/>A2A错误]
    JSONRPCError[JSONRPCError<br/>JSON-RPC错误]
    
    %% 事件
    TaskStatusUpdateEvent[TaskStatusUpdateEvent<br/>状态更新事件]
    TaskArtifactUpdateEvent[TaskArtifactUpdateEvent<br/>工件更新事件]
    
    %% 关系连接
    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
    
    %% 样式
    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

协议流程图

sequenceDiagram
    participant Client as 客户端
    participant Agent as 代理
    
    Note over Client,Agent: 1. 代理发现和能力查询
    Client->>Agent: GET /agent-card
    Agent->>Client: AgentCard（技能、能力、安全要求）
    
    Note over Client,Agent: 2. 认证（如果需要）
    Client->>Agent: 认证请求（根据SecurityScheme）
    Agent->>Client: 认证响应
    
    Note over Client,Agent: 3. 发送消息并创建任务
    Client->>Agent: SendMessageRequest
    Note right of Agent: 创建任务<br/>状态：submitted
    Agent->>Client: SendMessageResponse（Task）
    
    Note over Client,Agent: 4. 任务处理（可选推送通知）
    loop 任务处理
        Note right of Agent: 更新任务状态<br/>working -> completed
        alt 支持推送通知
            Agent->>Client: TaskStatusUpdateEvent
        else 轮询模式
            Client->>Agent: GetTaskRequest
            Agent->>Client: GetTaskResponse（更新的Task）
        end
    end
    
    Note over Client,Agent: 5. 获取最终结果
    Client->>Agent: GetTaskRequest
    Agent->>Client: Task（包含Artifacts和完整历史）

任务状态转换图

stateDiagram-v2
    [*] --> submitted: 创建任务
    
    submitted --> working: 开始处理
    submitted --> rejected: 拒绝任务
    submitted --> auth_required: 需要认证
    
    working --> completed: 处理完成
    working --> failed: 处理失败
    working --> input_required: 需要用户输入
    working --> canceled: 用户取消
    
    input_required --> working: 收到用户输入
    input_required --> canceled: 用户取消
    
    auth_required --> working: 认证成功
    auth_required --> rejected: 认证失败
    
    completed --> [*]
    failed --> [*]
    canceled --> [*]
    rejected --> [*]
    
    unknown --> working: 状态恢复
    unknown --> failed: 无法恢复

消息部分类型层次图

flowchart TD
    Part["Part<br/>消息部分基类"]
    
    Part --> TextPart["TextPart<br/>文本部分<br/>kind: text"]
    Part --> FilePart["FilePart<br/>文件部分<br/>kind: file"]
    Part --> DataPart["DataPart<br/>数据部分<br/>kind: data"]
    
    FilePart --> FileContent{"文件内容"}
    FileContent --> FileWithBytes["FileWithBytes<br/>包含base64字节数据"]
    FileContent --> FileWithUri["FileWithUri<br/>包含文件URI"]
    
    TextPart --> TextContent["text: str<br/>文本内容"]
    DataPart --> DataContent["data: dict[str, Any]<br/>结构化数据"]
    
    FileWithBytes --> BytesContent["bytes: str<br/>base64编码内容"]
    FileWithUri --> UriContent["uri: str<br/>文件URL"]
    
    %% 通用属性
    Part --> Metadata["metadata: dict[str, Any]<br/>可选元数据"]
    FileWithBytes --> FileMetadata["name: str<br/>mimeType: str<br/>文件元数据"]
    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

安全方案类型图

flowchart TD
    SecurityScheme["SecurityScheme<br/>安全方案联合类型"]
    
    SecurityScheme --> APIKeySecurityScheme["APIKeySecurityScheme<br/>API密钥认证<br/>type: apiKey"]
    SecurityScheme --> HTTPAuthSecurityScheme["HTTPAuthSecurityScheme<br/>HTTP认证<br/>type: http"]
    SecurityScheme --> OAuth2SecurityScheme["OAuth2SecurityScheme<br/>OAuth2认证<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

错误代码映射表

错误代码	错误类型	描述	使用场景
-32700	JSONParseError	JSON解析错误	无效的JSON格式
-32600	InvalidRequestError	无效请求	请求格式不符合规范
-32601	MethodNotFoundError	方法未找到	调用不存在的方法
-32602	InvalidParamsError	无效参数	方法参数不正确
-32603	InternalError	内部错误	服务器内部处理错误
-32001	TaskNotFoundError	任务未找到	请求的任务ID不存在
-32002	TaskNotCancelableError	任务不可取消	任务状态不允许取消
-32003	PushNotificationNotSupportedError	推送通知不支持	代理不支持推送通知
-32004	UnsupportedOperationError	操作不支持	代理不支持请求的操作
-32005	ContentTypeNotSupportedError	内容类型不支持	请求的内容类型与代理能力不匹配
-32006	InvalidAgentResponseError	无效代理响应	代理返回格式错误的响应

支持的方法列表

A2A协议定义了以下标准方法：

消息相关方法

message/send - 向代理发送消息
message/stream - 向代理发送流式消息

任务相关方法

tasks/get - 获取任务详情
tasks/cancel - 取消任务
tasks/resubscribe - 重新订阅任务更新

推送通知配置方法

tasks/pushNotificationConfig/set - 设置推送通知配置
tasks/pushNotificationConfig/get - 获取推送通知配置
tasks/pushNotificationConfig/list - 列出推送通知配置
tasks/pushNotificationConfig/delete - 删除推送通知配置

使用示例

1. 发送简单文本消息

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

# 创建文本消息部分
text_part = TextPart(text="你好，你能如何帮助我？")
part = Part(root=text_part)

# 创建消息
message = Message(
    messageId="msg-001",
    role="user",
    parts=[part]
)

# 创建发送参数
params = MessageSendParams(message=message)

# 创建请求
request = SendMessageRequest(
    id="req-001",
    params=params
)

2. 发送带文件的消息

from a2a.types import FilePart, FileWithBytes
import base64

# 读取文件并编码
with open("document.pdf", "rb") as f:
    file_bytes = base64.b64encode(f.read()).decode()

# 创建文件部分
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)

# 创建带文件的消息
message = Message(
    messageId="msg-002",
    role="user",
    parts=[part]
)

3. 查询任务状态

from a2a.types import GetTaskRequest, TaskQueryParams

# 创建任务查询请求
request = GetTaskRequest(
    id="req-002",
    params=TaskQueryParams(
        id="task-001",
        historyLength=10  # 获取最近10条消息历史
    )
)

4. 配置推送通知

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

# 创建推送通知配置
push_config = PushNotificationConfig(
    url="https://my-app.com/webhook/task-updates",
    authentication=PushNotificationAuthenticationInfo(
        schemes=["Bearer"],
        credentials="my-auth-token"
    )
)

# 创建任务推送配置
task_push_config = TaskPushNotificationConfig(
    taskId="task-001",
    pushNotificationConfig=push_config
)

# 创建设置请求
request = SetTaskPushNotificationConfigRequest(
    id="req-003",
    params=task_push_config
)

最佳实践

1. 错误处理

始终检查响应中的错误字段
根据错误代码实现适当的重试策略
为用户提供有意义的错误消息

2. 任务管理

使用推送通知而不是轮询来获取任务更新
实现任务超时机制
保存任务ID以便后续查询

3. 安全性

使用HTTPS传输
正确实现认证方案
定期轮换API密钥和令牌

4. 性能优化

使用流式传输处理大型响应
限制历史消息长度
实现客户端缓存

扩展性

A2A协议通过以下机制支持扩展：

代理扩展 - 通过AgentExtension声明自定义能力
元数据字段 - 大多数对象都包含可选的metadata字段
自定义传输协议 - 通过AgentInterface支持新的传输方法
扩展URI - 在消息和工件中引用扩展规范

总结

A2A协议为智能代理之间的通信和协作提供了一个全面的框架。通过标准化的消息格式、任务管理、安全认证和错误处理，该协议确保了不同代理实现之间的互操作性。

协议的设计考虑了扩展性和向后兼容性，使其能够适应未来的需求变化。通过支持多种内容类型、传输协议和认证方案，A2A协议为构建复杂的代理生态系统提供了坚实的基础。

A2A Specification

A2A协议规范（Python）

概述