Überblick

Das A2A (Agent2Agent) Protokoll ist ein auf JSON-RPC 2.0 basierendes Kommunikationsprotokoll, das für die Interaktion zwischen intelligenten Agenten entwickelt wurde. Dieses Protokoll definiert Kernfunktionalitäten einschließlich Agentenfähigkeitsdeklaration, Nachrichtenübertragung, Aufgabenverwaltung und Sicherheitsauthentifizierung.

Dieses Dokument basiert auf der Datei a2a-python/src/a2a/types.py und bietet eine detaillierte Einführung in alle Datenstrukturen und Objektbeziehungen des A2A-Protokolls.

Protokollversion

Aktuelle Protokollversion: 0.2.5

Kernkonzepte

1. Agent

Der Agent ist die Hauptentität des A2A-Protokolls und besitzt spezifische Fähigkeiten und Kompetenzen zur Bearbeitung von Benutzeranfragen und Ausführung von Aufgaben.

2. Aufgabe

Die Aufgabe ist die vom Agenten ausgeführte Arbeitseinheit, charakterisiert durch Lebenszyklus-Statusverwaltung und Verlaufsaufzeichnung.

3. Nachricht

Die Nachricht ist die grundlegende Einheit für den Informationsaustausch zwischen Benutzer und Agent und unterstützt mehrere Inhaltstypen.

4. JSON-RPC 2.0

Das A2A-Protokoll basiert auf dem JSON-RPC 2.0-Standard und bietet ein standardisiertes Anfrage-Antwort-Kommunikationsmuster.

Datenstruktur-Details

Grundtypen

A2A

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

Wurzeltyp des Protokolls, kann beliebige Daten enthalten.

Role (Aufzählung)

class Role(str, Enum):
    """Rolle des Nachrichtensenders"""
    agent = 'agent'  # Agent
    user = 'user'    # Benutzer

TaskState (Aufzählung)

class TaskState(str, Enum):
    """Mögliche Zustände der Aufgabe"""
    submitted = 'submitted'           # Eingereicht
    working = 'working'               # In Bearbeitung
    input_required = 'input-required' # Eingabe erforderlich
    completed = 'completed'           # Abgeschlossen
    canceled = 'canceled'             # Abgebrochen
    failed = 'failed'                 # Fehlgeschlagen
    rejected = 'rejected'             # Abgelehnt
    auth_required = 'auth-required'   # Authentifizierung erforderlich
    unknown = 'unknown'               # Unbekannter Status

Agentenbezogene Typen

AgentCard

class AgentCard(A2ABaseModel):
    """Agentenkarte - enthält wichtige Informationen des Agenten"""
    name: str                                    # Agentenname
    description: str                             # Agentenbeschreibung
    version: str                                 # Agentenversion
    url: str                                     # Agenten-URL
    protocolVersion: str | None = '0.2.5'       # Protokollversion
    skills: list[AgentSkill]                     # Fähigkeitenliste
    capabilities: AgentCapabilities              # Fähigkeitsdeklaration
    defaultInputModes: list[str]                 # Standard-Eingabemodi
    defaultOutputModes: list[str]                # Standard-Ausgabemodi
    provider: AgentProvider | None = None        # Dienstanbieter
    security: list[dict[str, list[str]]] | None = None  # Sicherheitsanforderungen
    securitySchemes: dict[str, SecurityScheme] | None = None  # Sicherheitsschemata
    # ... andere Felder

AgentSkill

class AgentSkill(A2ABaseModel):
    """Agentenfähigkeit - vom Agenten ausführbare Fähigkeitseinheit"""
    id: str                          # Eindeutige Fähigkeitskennung
    name: str                        # Fähigkeitsname
    description: str                 # Fähigkeitsbeschreibung
    tags: list[str]                  # Fähigkeits-Tags
    examples: list[str] | None = None # Verwendungsbeispiele
    inputModes: list[str] | None = None   # Eingabemodi
    outputModes: list[str] | None = None  # Ausgabemodi

AgentCapabilities

class AgentCapabilities(A2ABaseModel):
    """Agentenfähigkeitsdefinition"""
    extensions: list[AgentExtension] | None = None      # Unterstützte Erweiterungen
    pushNotifications: bool | None = None               # Push-Benachrichtigungsunterstützung
    stateTransitionHistory: bool | None = None          # Statusübergangsverlauf
    streaming: bool | None = None                       # Streaming-Unterstützung

AgentExtension

class AgentExtension(A2ABaseModel):
    """Agentenerweiterungsdeklaration"""
    uri: str                                # Erweiterungs-URI
    description: str | None = None          # Erweiterungsbeschreibung
    params: dict[str, Any] | None = None    # Erweiterungsparameter
    required: bool | None = None            # Erforderlich oder nicht

AgentProvider

class AgentProvider(A2ABaseModel):
    """Agenten-Dienstanbieter"""
    organization: str  # Organisationsname
    url: str          # Organisations-URL

AgentInterface

class AgentInterface(A2ABaseModel):
    """Agentenschnittstellendeklaration"""
    transport: str  # Transportprotokoll (JSONRPC, GRPC, HTTP+JSON)
    url: str       # Schnittstellen-URL

Nachrichtenbezogene Typen

Message

class Message(A2ABaseModel):
    """Zwischen Benutzer und Agent ausgetauschte Nachrichten"""
    messageId: str                           # Nachrichten-ID
    role: Role                               # Rolle des Senders
    parts: list[Part]                        # Nachrichteninhalts-Teile
    kind: Literal['message'] = 'message'     # Ereignistyp
    taskId: str | None = None                # Zugehörige Aufgaben-ID
    contextId: str | None = None             # Kontext-ID
    referenceTaskIds: list[str] | None = None # Referenzierte Aufgaben-ID-Liste
    extensions: list[str] | None = None       # Erweiterungs-URI-Liste
    metadata: dict[str, Any] | None = None    # Metadaten

Part (Union-Typ)

class Part(RootModel[TextPart | FilePart | DataPart]):
    """Nachrichtenteil - Text, Datei oder strukturierte Daten"""
    root: TextPart | FilePart | DataPart

TextPart

class TextPart(A2ABaseModel):
    """Text-Nachrichtenteil"""
    kind: Literal['text'] = 'text'           # Teiltyp
    text: str                                # Textinhalt
    metadata: dict[str, Any] | None = None   # Metadaten

FilePart

class FilePart(A2ABaseModel):
    """Datei-Nachrichtenteil"""
    kind: Literal['file'] = 'file'           # Teiltyp
    file: FileWithBytes | FileWithUri        # Dateiinhalt
    metadata: dict[str, Any] | None = None   # Metadaten

DataPart

class DataPart(A2ABaseModel):
    """Strukturierte Daten-Nachrichtenteil"""
    kind: Literal['data'] = 'data'           # Teiltyp
    data: dict[str, Any]                     # Strukturierte Daten
    metadata: dict[str, Any] | None = None   # Metadaten

Dateitypen

FileWithBytes

class FileWithBytes(A2ABaseModel):
    """Datei mit Byte-Inhalt"""
    bytes: str                        # base64-kodierter Dateiinhalt
    name: str | None = None           # Dateiname
    mimeType: str | None = None       # MIME-Typ

FileWithUri

class FileWithUri(A2ABaseModel):
    """Datei mit URI"""
    uri: str                          # Datei-URL
    name: str | None = None           # Dateiname
    mimeType: str | None = None       # MIME-Typ

Aufgabenbezogene Typen

Task

class Task(A2ABaseModel):
    """Aufgabenentität"""
    id: str                                  # Eindeutige Aufgabenkennung
    contextId: str                           # Kontext-ID
    status: TaskStatus                       # Aufgabenstatus
    kind: Literal['task'] = 'task'           # Ereignistyp
    history: list[Message] | None = None     # Nachrichtenverlauf
    artifacts: list[Artifact] | None = None  # Generierte Artefakte
    metadata: dict[str, Any] | None = None   # Metadaten

TaskStatus

class TaskStatus(A2ABaseModel):
    """Aufgabenstatus und zugehörige Nachricht"""
    state: TaskState                         # Aufgabenzustand
    message: Message | None = None           # Status-Update-Nachricht
    timestamp: str | None = None             # Zeitstempel (ISO 8601)

Artifact

class Artifact(A2ABaseModel):
    """Von der Aufgabe generiertes Artefakt"""
    artifactId: str                          # Artefakt-ID
    parts: list[Part]                        # Artefaktinhalts-Teile
    name: str | None = None                  # Artefaktname
    description: str | None = None           # Artefaktbeschreibung
    extensions: list[str] | None = None      # Erweiterungs-URI-Liste
    metadata: dict[str, Any] | None = None   # Metadaten

JSON-RPC-bezogene Typen

JSONRPCMessage (Basisklasse)

class JSONRPCMessage(A2ABaseModel):
    """JSON-RPC 2.0 Nachrichten-Basisklasse"""
    jsonrpc: Literal['2.0'] = '2.0'          # Protokollversion
    id: str | int | None = None              # Anfrage/Antwort-ID

JSONRPCRequest

class JSONRPCRequest(A2ABaseModel):
    """JSON-RPC 2.0 Anfrageobjekt"""
    jsonrpc: Literal['2.0'] = '2.0'          # Protokollversion
    method: str                              # Methodenname
    params: dict[str, Any] | None = None     # Parameter
    id: str | int | None = None              # Anfrage-ID

JSONRPCSuccessResponse

class JSONRPCSuccessResponse(A2ABaseModel):
    """JSON-RPC 2.0 Erfolgsantwortobjekt"""
    jsonrpc: Literal['2.0'] = '2.0'          # Protokollversion
    result: Any                              # Ergebnisdaten
    id: str | int | None = None              # Antwort-ID

JSONRPCErrorResponse

class JSONRPCErrorResponse(A2ABaseModel):
    """JSON-RPC 2.0 Fehlerantwortobjekt"""
    jsonrpc: Literal['2.0'] = '2.0'          # Protokollversion
    error: JSONRPCError | [specific error type]      # Fehlerinformationen
    id: str | int | None = None              # Antwort-ID

Fehlertypen

Das A2A-Protokoll definiert mehrere Fehlertypen, alle von Standard-JSON-RPC-Fehlern abgeleitet:

Standard-JSON-RPC-Fehler

JSONParseError

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

InvalidRequestError

class InvalidRequestError(A2ABaseModel):
    """Ungültige Anfragefehler"""
    code: Literal[-32600] = -32600
    message: str | None = 'Request payload validation error'
    data: Any | None = None

MethodNotFoundError

class MethodNotFoundError(A2ABaseModel):
    """Methode nicht gefunden Fehler"""
    code: Literal[-32601] = -32601
    message: str | None = 'Method not found'
    data: Any | None = None

InvalidParamsError

class InvalidParamsError(A2ABaseModel):
    """Ungültige Parameter Fehler"""
    code: Literal[-32602] = -32602
    message: str | None = 'Invalid parameters'
    data: Any | None = None

InternalError

class InternalError(A2ABaseModel):
    """Interner Fehler"""
    code: Literal[-32603] = -32603
    message: str | None = 'Internal error'
    data: Any | None = None

A2A-spezifische Fehler

TaskNotFoundError

class TaskNotFoundError(A2ABaseModel):
    """Aufgabe nicht gefunden Fehler"""
    code: Literal[-32001] = -32001
    message: str | None = 'Task not found'
    data: Any | None = None

TaskNotCancelableError

class TaskNotCancelableError(A2ABaseModel):
    """Aufgabe nicht abbrechbar Fehler"""
    code: Literal[-32002] = -32002
    message: str | None = 'Task cannot be canceled'
    data: Any | None = None

PushNotificationNotSupportedError

class PushNotificationNotSupportedError(A2ABaseModel):
    """Push-Benachrichtigung nicht unterstützt Fehler"""
    code: Literal[-32003] = -32003
    message: str | None = 'Push Notification is not supported'
    data: Any | None = None

UnsupportedOperationError

class UnsupportedOperationError(A2ABaseModel):
    """Nicht unterstützte Operation Fehler"""
    code: Literal[-32004] = -32004
    message: str | None = 'This operation is not supported'
    data: Any | None = None

ContentTypeNotSupportedError

class ContentTypeNotSupportedError(A2ABaseModel):
    """Inhaltstyp nicht unterstützt Fehler"""
    code: Literal[-32005] = -32005
    message: str | None = 'Incompatible content types'
    data: Any | None = None

InvalidAgentResponseError

class InvalidAgentResponseError(A2ABaseModel):
    """Ungültige Agentenantwort Fehler"""
    code: Literal[-32006] = -32006
    message: str | None = 'Invalid agent response'
    data: Any | None = None

Sicherheitsauthentifizierungsbezogene Typen

SecurityScheme (Union-Typ)

class SecurityScheme(RootModel[
    APIKeySecurityScheme |
    HTTPAuthSecurityScheme |
    OAuth2SecurityScheme |
    OpenIdConnectSecurityScheme
]):
    """Sicherheitsschema - unterstützt mehrere Authentifizierungsmethoden"""

APIKeySecurityScheme

class APIKeySecurityScheme(A2ABaseModel):
    """API-Schlüssel-Sicherheitsschema"""
    type: Literal['apiKey'] = 'apiKey'
    name: str                              # Parametername
    in_: In                                # Ort (header/query/cookie)
    description: str | None = None         # Beschreibung

HTTPAuthSecurityScheme

class HTTPAuthSecurityScheme(A2ABaseModel):
    """HTTP-Authentifizierungs-Sicherheitsschema"""
    type: Literal['http'] = 'http'
    scheme: str                            # Authentifizierungsschema (Basic, Bearer usw.)
    bearerFormat: str | None = None        # Bearer-Token-Format
    description: str | None = None         # Beschreibung

OAuth2SecurityScheme

class OAuth2SecurityScheme(A2ABaseModel):
    """OAuth2.0-Sicherheitsschema"""
    type: Literal['oauth2'] = 'oauth2'
    flows: OAuthFlows                      # OAuth-Flow-Einstellungen
    description: str | None = None         # Beschreibung

OpenIdConnectSecurityScheme

class OpenIdConnectSecurityScheme(A2ABaseModel):
    """OpenID Connect-Sicherheitsschema"""
    type: Literal['openIdConnect'] = 'openIdConnect'
    openIdConnectUrl: str                  # OpenID Connect-Discovery-URL
    description: str | None = None         # Beschreibung

OAuth-Flow-Typen

OAuthFlows

class OAuthFlows(A2ABaseModel):
    """OAuth-Flow-Einstellungen"""
    implicit: ImplicitOAuthFlow | None = None                    # Impliziter Flow
    password: PasswordOAuthFlow | None = None                    # Passwort-Flow
    clientCredentials: ClientCredentialsOAuthFlow | None = None  # Client-Credentials-Flow
    authorizationCode: AuthorizationCodeOAuthFlow | None = None  # Authorization-Code-Flow

Push-Benachrichtigungsbezogene Typen

PushNotificationConfig

class PushNotificationConfig(A2ABaseModel):
    """Push-Benachrichtigungskonfiguration"""
    url: str                                                    # Push-URL
    id: str | None = None                                       # Push-Benachrichtigungs-ID
    token: str | None = None                                    # Sitzungstoken
    authentication: PushNotificationAuthenticationInfo | None = None  # Authentifizierungsinformationen

PushNotificationAuthenticationInfo

class PushNotificationAuthenticationInfo(A2ABaseModel):
    """Push-Benachrichtigungs-Authentifizierungsinformationen"""
    schemes: list[str]                     # Unterstützte Authentifizierungsschemata
    credentials: str | None = None         # Authentifizierungsinformationen

Anfrage- und Antworttypen

Nachrichtensendung

SendMessageRequest

class SendMessageRequest(A2ABaseModel):
    """Nachrichtensendungsanfrage"""
    jsonrpc: Literal['2.0'] = '2.0'
    method: Literal['message/send'] = 'message/send'
    params: MessageSendParams
    id: str | int

MessageSendParams

class MessageSendParams(A2ABaseModel):
    """Nachrichtensendungsparameter"""
    message: Message                                    # Zu sendende Nachricht
    configuration: MessageSendConfiguration | None = None  # Sendungseinstellungen
    metadata: dict[str, Any] | None = None             # Metadaten

MessageSendConfiguration

class MessageSendConfiguration(A2ABaseModel):
    """Nachrichtensendungskonfiguration"""
    acceptedOutputModes: list[str]                          # Akzeptierte Ausgabemodi
    blocking: bool | None = None                            # Anfrage blockieren oder nicht
    historyLength: int | None = None                        # Nachrichtenverlaufslänge
    pushNotificationConfig: PushNotificationConfig | None = None  # Push-Benachrichtigungseinstellungen

Aufgabenoperationen

GetTaskRequest

class GetTaskRequest(A2ABaseModel):
    """Aufgabenabrufanfrage"""
    jsonrpc: Literal['2.0'] = '2.0'
    method: Literal['tasks/get'] = 'tasks/get'
    params: TaskQueryParams
    id: str | int

TaskQueryParams

class TaskQueryParams(A2ABaseModel):
    """Aufgabenabfrageparameter"""
    id: str                              # Aufgaben-ID
    historyLength: int | None = None     # Verlaufslänge
    metadata: dict[str, Any] | None = None  # Metadaten

CancelTaskRequest

class CancelTaskRequest(A2ABaseModel):
    """Aufgabenabbruchanfrage"""
    jsonrpc: Literal['2.0'] = '2.0'
    method: Literal['tasks/cancel'] = 'tasks/cancel'
    params: TaskIdParams
    id: str | int

Ereignistypen

TaskStatusUpdateEvent

class TaskStatusUpdateEvent(A2ABaseModel):
    """Aufgabenstatus-Update-Ereignis"""
    kind: Literal['status-update'] = 'status-update'
    taskId: str                              # Aufgaben-ID
    contextId: str                           # Kontext-ID
    status: TaskStatus                       # Aufgabenstatus
    final: bool                              # Finales Ereignis oder nicht
    metadata: dict[str, Any] | None = None   # Metadaten

TaskArtifactUpdateEvent

class TaskArtifactUpdateEvent(A2ABaseModel):
    """Aufgabenartefakt-Update-Ereignis"""
    kind: Literal['artifact-update'] = 'artifact-update'
    taskId: str                              # Aufgaben-ID
    contextId: str                           # Kontext-ID
    artifact: Artifact                       # Artefakt
    append: bool | None = None               # Anhängen oder nicht
    lastChunk: bool | None = None            # Letzter Chunk oder nicht
    metadata: dict[str, Any] | None = None   # Metadaten

Objektbeziehungsdiagramm

graph TB
    %% Hauptentitäten
    AgentCard[AgentCard<br/>Agentenkarte]
    Task[Task<br/>Aufgabe]
    Message[Message<br/>Nachricht]
    
    %% Agentenbezogen
    AgentSkill[AgentSkill<br/>Agentenfähigkeit]
    AgentCapabilities[AgentCapabilities<br/>Agentenfähigkeiten]
    AgentProvider[AgentProvider<br/>Dienstanbieter]
    AgentExtension[AgentExtension<br/>Agentenerweiterung]
    AgentInterface[AgentInterface<br/>Agentenschnittstelle]
    
    %% Nachrichteninhalt
    Part[Part<br/>Nachrichtenteil]
    TextPart[TextPart<br/>Textteil]
    FilePart[FilePart<br/>Dateiteil]
    DataPart[DataPart<br/>Datenteil]
    
    %% Dateitypen
    FileWithBytes[FileWithBytes<br/>Byte-Datei]
    FileWithUri[FileWithUri<br/>URI-Datei]
    
    %% Aufgabenbezogen
    TaskStatus[TaskStatus<br/>Aufgabenstatus]
    Artifact[Artifact<br/>Artefakt]
    
    %% JSON-RPC
    JSONRPCRequest[JSONRPCRequest<br/>JSON-RPC-Anfrage]
    JSONRPCResponse[JSONRPCResponse<br/>JSON-RPC-Antwort]
    
    %% Sicherheitsauthentifizierung
    SecurityScheme[SecurityScheme<br/>Sicherheitsschema]
    APIKeySecurityScheme[APIKeySecurityScheme<br/>API-Schlüssel-Schema]
    HTTPAuthSecurityScheme[HTTPAuthSecurityScheme<br/>HTTP-Auth-Schema]
    OAuth2SecurityScheme[OAuth2SecurityScheme<br/>OAuth2-Schema]
    OpenIdConnectSecurityScheme[OpenIdConnectSecurityScheme<br/>OpenID Connect-Schema]
    
    %% Push-Benachrichtigungen
    PushNotificationConfig[PushNotificationConfig<br/>Push-Benachrichtigungskonfiguration]
    PushNotificationAuthenticationInfo[PushNotificationAuthenticationInfo<br/>Push-Auth-Informationen]
    
    %% Fehlertypen
    A2AError[A2AError<br/>A2A-Fehler]
    JSONRPCError[JSONRPCError<br/>JSON-RPC-Fehler]
    
    %% Ereignisse
    TaskStatusUpdateEvent[TaskStatusUpdateEvent<br/>Status-Update-Ereignis]
    TaskArtifactUpdateEvent[TaskArtifactUpdateEvent<br/>Artefakt-Update-Ereignis]
    
    %% Beziehungsverbindungen
    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
    
    %% Stile
    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

Protokoll-Flussdiagramm

sequenceDiagram
    participant Client as Client
    participant Agent as Agent
    
    Note over Client,Agent: 1. Agentenerkennung und Fähigkeitsabfrage
    Client->>Agent: GET /agent-card
    Agent->>Client: AgentCard (Fähigkeiten, Kompetenzen, Sicherheitsanforderungen)
    
    Note over Client,Agent: 2. Authentifizierung (falls erforderlich)
    Client->>Agent: Authentifizierungsanfrage (gemäß SecurityScheme)
    Agent->>Client: Authentifizierungsantwort
    
    Note over Client,Agent: 3. Nachrichtensendung und Aufgabenerstellung
    Client->>Agent: SendMessageRequest
    Note right of Agent: Aufgabenerstellung<br/>Status: submitted
    Agent->>Client: SendMessageResponse (Task)
    
    Note over Client,Agent: 4. Aufgabenverarbeitung (optionale Push-Benachrichtigung)
    loop Aufgabenverarbeitung
        Note right of Agent: Aufgabenstatus-Update<br/>working -> completed
        alt Push-Benachrichtigungsunterstützung
            Agent->>Client: TaskStatusUpdateEvent
        else Polling-Modus
            Client->>Agent: GetTaskRequest
            Agent->>Client: GetTaskResponse (aktualisierte Task)
        end
    end
    
    Note over Client,Agent: 5. Finales Ergebnis abrufen
    Client->>Agent: GetTaskRequest
    Agent->>Client: Task (mit Artifacts und vollständigem Verlauf)

Aufgabenstatus-Übergangsdiagramm

stateDiagram-v2
    [*] --> submitted: Aufgabenerstellung
    
    submitted --> working: Verarbeitung beginnt
    submitted --> rejected: Aufgabe abgelehnt
    submitted --> auth_required: Authentifizierung erforderlich
    
    working --> completed: Verarbeitung abgeschlossen
    working --> failed: Verarbeitung fehlgeschlagen
    working --> input_required: Benutzereingabe erforderlich
    working --> canceled: Benutzerabbruch
    
    input_required --> working: Benutzereingabe erhalten
    input_required --> canceled: Benutzerabbruch
    
    auth_required --> working: Authentifizierung erfolgreich
    auth_required --> rejected: Authentifizierung fehlgeschlagen
    
    completed --> [*]
    failed --> [*]
    canceled --> [*]
    rejected --> [*]
    
    unknown --> working: Statuswiederherstellung
    unknown --> failed: Wiederherstellung unmöglich

Nachrichtenteil-Typ-Hierarchiediagramm

flowchart TD
    Part["Part<br/>Nachrichtenteil-Basisklasse"]
    
    Part --> TextPart["TextPart<br/>Textteil<br/>kind: text"]
    Part --> FilePart["FilePart<br/>Dateiteil<br/>kind: file"]
    Part --> DataPart["DataPart<br/>Datenteil<br/>kind: data"]
    
    FilePart --> FileContent{"Dateiinhalt"}
    FileContent --> FileWithBytes["FileWithBytes<br/>Enthält base64-Byte-Daten"]
    FileContent --> FileWithUri["FileWithUri<br/>Enthält Datei-URI"]
    
    TextPart --> TextContent["text: str<br/>Textinhalt"]
    DataPart --> DataContent["data: dict[str, Any]<br/>Strukturierte Daten"]
    
    FileWithBytes --> BytesContent["bytes: str<br/>base64-kodierter Inhalt"]
    FileWithUri --> UriContent["uri: str<br/>Datei-URL"]
    
    %% Gemeinsame Eigenschaften
    Part --> Metadata["metadata: dict[str, Any]<br/>Optionale Metadaten"]
    FileWithBytes --> FileMetadata["name: str<br/>mimeType: str<br/>Dateimetadaten"]
    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

Sicherheitsschema-Typdiagramm

flowchart TD
    SecurityScheme["SecurityScheme<br/>Sicherheitsschema-Union-Typ"]
    
    SecurityScheme --> APIKeySecurityScheme["APIKeySecurityScheme<br/>API-Schlüssel-Authentifizierung<br/>type: apiKey"]
    SecurityScheme --> HTTPAuthSecurityScheme["HTTPAuthSecurityScheme<br/>HTTP-Authentifizierung<br/>type: http"]
    SecurityScheme --> OAuth2SecurityScheme["OAuth2SecurityScheme<br/>OAuth2-Authentifizierung<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

Fehlercode-Zuordnungstabelle

Liste unterstützter Methoden

Das A2A-Protokoll definiert die folgenden Standardmethoden:

Nachrichtenbezogene Methoden

• message/send - Nachricht an Agent senden
• message/stream - Streaming-Nachricht an Agent senden

Aufgabenbezogene Methoden

• tasks/get - Aufgabendetails abrufen
• tasks/cancel - Aufgabe abbrechen
• tasks/resubscribe - Aufgaben-Updates erneut abonnieren

Push-Benachrichtigungskonfigurationsmethoden

• tasks/pushNotificationConfig/set - Push-Benachrichtigungskonfiguration festlegen
• tasks/pushNotificationConfig/get - Push-Benachrichtigungskonfiguration abrufen
• tasks/pushNotificationConfig/list - Push-Benachrichtigungskonfigurationen auflisten
• tasks/pushNotificationConfig/delete - Push-Benachrichtigungskonfiguration löschen

Verwendungsbeispiele

1. Einfache Textnachricht senden

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

# Textnachrichtenteil erstellen
text_part = TextPart(text="Hallo, wie kann ich Ihnen helfen?")
part = Part(root=text_part)

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

# Sendungsparameter erstellen
params = MessageSendParams(message=message)

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

2. Nachricht mit Datei senden

from a2a.types import FilePart, FileWithBytes
import base64

# Datei lesen und kodieren
with open("document.pdf", "rb") as f:
    file_bytes = base64.b64encode(f.read()).decode()

# Dateiteil erstellen
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)

# Nachricht mit Datei erstellen
message = Message(
    messageId="msg-002",
    role="user",
    parts=[part]
)

3. Aufgabenstatus-Abfrage

from a2a.types import GetTaskRequest, TaskQueryParams

# Aufgabenabfrageanfrage erstellen
request = GetTaskRequest(
    id="req-002",
    params=TaskQueryParams(
        id="task-001",
        historyLength=10  # Neueste 10 Nachrichtenverlauf abrufen
    )
)

4. Push-Benachrichtigungskonfiguration

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

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

# Aufgaben-Push-Konfiguration erstellen
task_push_config = TaskPushNotificationConfig(
    taskId="task-001",
    pushNotificationConfig=push_config
)

# Konfigurationsanfrage erstellen
request = SetTaskPushNotificationConfigRequest(
    id="req-003",
    params=task_push_config
)

Best Practices

1. Fehlerbehandlung

• Immer das Fehlerfeld der Antwort überprüfen
• Angemessene Wiederholungsstrategie basierend auf Fehlercode implementieren
• Aussagekräftige Fehlermeldungen für Benutzer bereitstellen

2. Aufgabenverwaltung

• Push-Benachrichtigungen anstelle von Polling für Aufgaben-Updates verwenden
• Aufgaben-Timeout-Mechanismus implementieren
• Aufgaben-IDs für spätere Abfragen speichern

3. Sicherheit

• HTTPS-Transport verwenden
• Authentifizierungsschemata ordnungsgemäß implementieren
• API-Schlüssel und Token regelmäßig rotieren

4. Leistungsoptimierung

• Streaming für große Antwortverarbeitung verwenden
• Nachrichtenverlaufslänge begrenzen
• Client-seitiges Caching implementieren

Erweiterbarkeit

Das A2A-Protokoll unterstützt Erweiterungen durch die folgenden Mechanismen:

1. Agentenerweiterungen - Benutzerdefinierte Fähigkeiten über AgentExtension deklarieren
2. Metadatenfelder - Die meisten Objekte enthalten ein optionales metadata-Feld
3. Benutzerdefinierte Transportprotokolle - Unterstützung neuer Transportmethoden über AgentInterface
4. Erweiterungs-URIs - Referenzierung von Erweiterungsspezifikationen in Nachrichten und Artefakten

Zusammenfassung

Das A2A-Protokoll bietet ein umfassendes Framework für Kommunikation und Zusammenarbeit zwischen intelligenten Agenten. Basierend auf JSON-RPC 2.0 bietet es standardisierte Nachrichtenübertragung, Aufgabenverwaltung, Sicherheitsauthentifizierung und Erweiterungsmechanismen.

Hauptmerkmale:

• Standardisierte Kommunikation: JSON-RPC 2.0-basiertes Anfrage-Antwort-Muster
• Reiche Inhaltsunterstützung: Unterstützung für Text, Dateien und strukturierte Daten
• Flexible Sicherheit: Unterstützung mehrerer Authentifizierungsmethoden
• Echtzeit-Updates: Push-Benachrichtigungsmechanismus
• Robuste Fehlerbehandlung: Umfassende Fehlercodes und Nachrichten
• Gute Erweiterbarkeit: Erweiterungsmechanismen und Metadatenunterstützung

Dieses Protokoll bietet Entwicklern eine solide Grundlage für den Aufbau leistungsstarker agentenbasierter Anwendungen.