From 10a037b737e0c48e4d850818ea15947c6682e33d Mon Sep 17 00:00:00 2001 From: brucearctor <5032356+brucearctor@users.noreply.github.com> Date: Thu, 26 Mar 2026 10:59:30 -0700 Subject: [PATCH 1/7] feat: add proto2pydantic Pydantic model generation pipeline Add protoc-gen-proto2pydantic plugin to buf.gen.yaml for generating Pydantic models alongside existing protobuf _pb2 objects. Generated models: - 39 Pydantic models + 2 enums from a2a.proto - All extend A2ABaseModel with to_proto_json() for ProtoJSON compat - Keyword escaping (list -> list_ with alias='list') - Timestamp field serializers for RFC 3339 - Oneof unions for Part.content, SecurityScheme.scheme, etc. This is an additive change - existing a2a_pb2 imports are untouched. New Pydantic models available via: from a2a.types.a2a_pydantic import ... --- buf.gen.yaml | 7 + scripts/gen_proto.sh | 4 + src/a2a/types/a2a_pydantic.py | 706 ++++++++++++++++++++++++++++++++++ 3 files changed, 717 insertions(+) create mode 100644 src/a2a/types/a2a_pydantic.py diff --git a/buf.gen.yaml b/buf.gen.yaml index d7937469c..37333a2da 100644 --- a/buf.gen.yaml +++ b/buf.gen.yaml @@ -29,3 +29,10 @@ plugins: # Generates *_pb2.pyi files. - remote: buf.build/protocolbuffers/pyi out: src/a2a/types + # Generate Pydantic models with ProtoJSON compatibility + - local: protoc-gen-proto2pydantic + out: src/a2a/types + opt: + - preset=a2a + - base_class=a2a._base.A2ABaseModel + - output_file=a2a_pydantic.py diff --git a/scripts/gen_proto.sh b/scripts/gen_proto.sh index 34ff96ae0..828f5322a 100755 --- a/scripts/gen_proto.sh +++ b/scripts/gen_proto.sh @@ -1,6 +1,10 @@ #!/bin/bash set -e +# Install proto2pydantic plugin for Pydantic model generation +echo "Installing protoc-gen-proto2pydantic@v0.4.0..." +go install github.com/protocgen/proto2pydantic@v0.4.0 + # Run buf generate to regenerate protobuf code and OpenAPI spec npx --yes @bufbuild/buf generate diff --git a/src/a2a/types/a2a_pydantic.py b/src/a2a/types/a2a_pydantic.py new file mode 100644 index 000000000..b1165ccbb --- /dev/null +++ b/src/a2a/types/a2a_pydantic.py @@ -0,0 +1,706 @@ +"""Generated by proto2pydantic from a2a.proto. DO NOT EDIT.""" + +from __future__ import annotations + +from datetime import datetime +from enum import Enum + +from typing import Any + +from a2a._base import A2ABaseModel +from pydantic import Field, field_serializer + + +class TaskState(str, Enum): + """Defines the possible lifecycle states of a `Task`.""" + + TASK_STATE_UNSPECIFIED = 'TASK_STATE_UNSPECIFIED' + TASK_STATE_SUBMITTED = 'TASK_STATE_SUBMITTED' + TASK_STATE_WORKING = 'TASK_STATE_WORKING' + TASK_STATE_COMPLETED = 'TASK_STATE_COMPLETED' + TASK_STATE_FAILED = 'TASK_STATE_FAILED' + TASK_STATE_CANCELED = 'TASK_STATE_CANCELED' + TASK_STATE_INPUT_REQUIRED = 'TASK_STATE_INPUT_REQUIRED' + TASK_STATE_REJECTED = 'TASK_STATE_REJECTED' + TASK_STATE_AUTH_REQUIRED = 'TASK_STATE_AUTH_REQUIRED' + + +class Role(str, Enum): + """Defines the sender of a message in A2A protocol communication.""" + + ROLE_UNSPECIFIED = 'ROLE_UNSPECIFIED' + ROLE_USER = 'ROLE_USER' + ROLE_AGENT = 'ROLE_AGENT' + + +class AuthenticationInfo(A2ABaseModel): + """Defines authentication details, used for push notifications.""" + + scheme: str = Field(..., description='HTTP Authentication Scheme from the [IANA registry](https://www.iana.org/assignments/http-authschemes/). Examples: `Bearer`, `Basic`, `Digest`. Scheme names are case-insensitive per [RFC 9110 Section 11.1](https://www.rfc-editor.org/rfc/rfc9110#section-11.1).') + credentials: str = Field(default='', description='Push Notification credentials. Format depends on the scheme (e.g., token for Bearer).') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class TaskPushNotificationConfig(A2ABaseModel): + """A container associating a push notification configuration with a specific task.""" + + tenant: str = Field(default='', description='Optional. Tenant ID.') + id: str = Field(default='', description='The push notification configuration details. A unique identifier (e.g. UUID) for this push notification configuration.') + task_id: str = Field(default='', description='The ID of the task this configuration is associated with.') + url: str = Field(..., description='The URL where the notification should be sent.') + token: str = Field(default='', description='A token unique for this task or session.') + authentication: AuthenticationInfo = Field(default=None, description='Authentication information required to send the notification.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class SendMessageConfiguration(A2ABaseModel): + """Configuration of a send message request.""" + + accepted_output_modes: list[str] | None = Field(default=None, description='A list of media types the client is prepared to accept for response parts. Agents SHOULD use this to tailor their output.') + task_push_notification_config: TaskPushNotificationConfig = Field(default=None, description='Configuration for the agent to send push notifications for task updates. Task id should be empty when sending this configuration in a `SendMessage` request.') + history_length: int | None = Field(default=None, description='The maximum number of most recent messages from the task\'s history to retrieve in the response. An unset value means the client does not impose any limit. A value of zero is a request to not include any messages. The server MUST NOT return more messages than the provided value, but MAY apply a lower limit.') + return_immediately: bool = Field(default=False, description='If `true`, the operation returns immediately after creating the task, even if processing is still in progress. If `false` (default), the operation MUST wait until the task reaches a terminal (`COMPLETED`, `FAILED`, `CANCELED`, `REJECTED`) or interrupted (`INPUT_REQUIRED`, `AUTH_REQUIRED`) state before returning.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class Part(A2ABaseModel): + """`Part` represents a container for a section of communication content. Parts can be purely textual, some sort of file (image, video, etc) or a structured data blob (i.e. JSON).""" + + metadata: dict[str, Any] = Field(default=None, description='Optional. metadata associated with this part.') + filename: str = Field(default='', description='An optional `filename` for the file (e.g., "document.pdf").') + media_type: str = Field(default='', description='The `media_type` (MIME type) of the part content (e.g., "text/plain", "application/json", "image/png"). This field is available for all part types.') + content: str | bytes | str | Any | None = None + + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class Message(A2ABaseModel): + """`Message` is one unit of communication between client and server. It can be associated with a context and/or a task. For server messages, `context_id` must be provided, and `task_id` only if a task was created. For client messages, both fields are optional, with the caveat that if both are provided, they have to match (the `context_id` has to be the one that is set on the task). If only `task_id` is provided, the server will infer `context_id` from it.""" + + message_id: str = Field(..., description='The unique identifier (e.g. UUID) of the message. This is created by the message creator.') + context_id: str = Field(default='', description='Optional. The context id of the message. If set, the message will be associated with the given context.') + task_id: str = Field(default='', description='Optional. The task id of the message. If set, the message will be associated with the given task.') + role: Role = Field(..., description='Identifies the sender of the message.') + parts: list[Part] = Field(..., description='Parts is the container of the message content.') + metadata: dict[str, Any] = Field(default=None, description='Optional. Any metadata to provide along with the message.') + extensions: list[str] | None = Field(default=None, description='The URIs of extensions that are present or contributed to this Message.') + reference_task_ids: list[str] | None = Field(default=None, description='A list of task IDs that this message references for additional context.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class TaskStatus(A2ABaseModel): + """A container for the status of a task""" + + state: TaskState = Field(..., description='The current state of this task.') + message: Message = Field(default=None, description='A message associated with the status.') + timestamp: datetime = Field(default=None, description='ISO 8601 Timestamp when the status was recorded. Example: "2023-10-27T10:00:00Z"') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + @field_serializer('timestamp') + @classmethod + def _serialize_timestamp(cls, v: datetime | None, _info: Any) -> str | None: + """Serialize datetime to RFC 3339 with UTC 'Z' suffix for ProtoJSON.""" + if v is None: + return None + return v.strftime('%Y-%m-%dT%H:%M:%S.') + f'{v.microsecond:06d}'[:3] + 'Z' + + + +class Task(A2ABaseModel): + """`Task` is the core unit of action for A2A. It has a current status and when results are created for the task they are stored in the artifact. If there are multiple turns for a task, these are stored in history.""" + + id: str = Field(..., description='Unique identifier (e.g. UUID) for the task, generated by the server for a new task.') + context_id: str = Field(default='', description='Unique identifier (e.g. UUID) for the contextual collection of interactions (tasks and messages).') + status: TaskStatus = Field(..., description='The current status of a `Task`, including `state` and a `message`.') + artifacts: list[Artifact] | None = Field(default=None, description='A set of output artifacts for a `Task`.') + history: list[Message] | None = Field(default=None, description='protolint:disable REPEATED_FIELD_NAMES_PLURALIZED The history of interactions from a `Task`.') + metadata: dict[str, Any] = Field(default=None, description='protolint:enable REPEATED_FIELD_NAMES_PLURALIZED A key/value object to store custom metadata about a task.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class Artifact(A2ABaseModel): + """Artifacts represent task outputs.""" + + artifact_id: str = Field(..., description='Unique identifier (e.g. UUID) for the artifact. It must be unique within a task.') + name: str = Field(default='', description='A human readable name for the artifact.') + description: str = Field(default='', description='Optional. A human readable description of the artifact.') + parts: list[Part] = Field(..., description='The content of the artifact. Must contain at least one part.') + metadata: dict[str, Any] = Field(default=None, description='Optional. Metadata included with the artifact.') + extensions: list[str] | None = Field(default=None, description='The URIs of extensions that are present or contributed to this Artifact.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class TaskStatusUpdateEvent(A2ABaseModel): + """An event sent by the agent to notify the client of a change in a task's status.""" + + task_id: str = Field(..., description='The ID of the task that has changed.') + context_id: str = Field(..., description='The ID of the context that the task belongs to.') + status: TaskStatus = Field(..., description='The new status of the task.') + metadata: dict[str, Any] = Field(default=None, description='Optional. Metadata associated with the task update.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class TaskArtifactUpdateEvent(A2ABaseModel): + """A task delta where an artifact has been generated.""" + + task_id: str = Field(..., description='The ID of the task for this artifact.') + context_id: str = Field(..., description='The ID of the context that this task belongs to.') + artifact: Artifact = Field(..., description='The artifact that was generated or updated.') + append: bool = Field(default=False, description='If true, the content of this artifact should be appended to a previously sent artifact with the same ID.') + last_chunk: bool = Field(default=False, description='If true, this is the final chunk of the artifact.') + metadata: dict[str, Any] = Field(default=None, description='Optional. Metadata associated with the artifact update.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class AgentInterface(A2ABaseModel): + """Declares a combination of a target URL, transport and protocol version for interacting with the agent. This allows agents to expose the same functionality over multiple protocol binding mechanisms.""" + + url: str = Field(..., description='The URL where this interface is available. Must be a valid absolute HTTPS URL in production. Example: "https://api.example.com/a2a/v1", "https://grpc.example.com/a2a"') + protocol_binding: str = Field(..., description='The protocol binding supported at this URL. This is an open form string, to be easily extended for other protocol bindings. The core ones officially supported are `JSONRPC`, `GRPC` and `HTTP+JSON`.') + tenant: str = Field(default='', description='Tenant ID to be used in the request when calling the agent.') + protocol_version: str = Field(..., description='The version of the A2A protocol this interface exposes. Use the latest supported minor version per major version. Examples: "0.3", "1.0"') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class AgentProvider(A2ABaseModel): + """Represents the service provider of an agent.""" + + url: str = Field(..., description='A URL for the agent provider\'s website or relevant documentation. Example: "https://ai.google.dev"') + organization: str = Field(..., description='The name of the agent provider\'s organization. Example: "Google"') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class AgentCapabilities(A2ABaseModel): + """Defines optional capabilities supported by an agent.""" + + streaming: bool | None = Field(default=None, description='Indicates if the agent supports streaming responses.') + push_notifications: bool | None = Field(default=None, description='Indicates if the agent supports sending push notifications for asynchronous task updates.') + extensions: list[AgentExtension] | None = Field(default=None, description='A list of protocol extensions supported by the agent.') + extended_agent_card: bool | None = Field(default=None, description='Indicates if the agent supports providing an extended agent card when authenticated.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class AgentSkill(A2ABaseModel): + """Represents a distinct capability or function that an agent can perform.""" + + id: str = Field(..., description='A unique identifier for the agent\'s skill.') + name: str = Field(..., description='A human-readable name for the skill.') + description: str = Field(..., description='A detailed description of the skill.') + tags: list[str] = Field(..., description='A set of keywords describing the skill\'s capabilities.') + examples: list[str] | None = Field(default=None, description='Example prompts or scenarios that this skill can handle.') + input_modes: list[str] | None = Field(default=None, description='The set of supported input media types for this skill, overriding the agent\'s defaults.') + output_modes: list[str] | None = Field(default=None, description='The set of supported output media types for this skill, overriding the agent\'s defaults.') + security_requirements: list[SecurityRequirement] | None = Field(default=None, description='Security schemes necessary for this skill.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class AgentCard(A2ABaseModel): + """A self-describing manifest for an agent. It provides essential metadata including the agent's identity, capabilities, skills, supported communication methods, and security requirements. Next ID: 20""" + + name: str = Field(..., description='A human readable name for the agent. Example: "Recipe Agent"') + description: str = Field(..., description='A human-readable description of the agent, assisting users and other agents in understanding its purpose. Example: "Agent that helps users with recipes and cooking."') + supported_interfaces: list[AgentInterface] = Field(..., description='Ordered list of supported interfaces. The first entry is preferred.') + provider: AgentProvider = Field(default=None, description='The service provider of the agent.') + version: str = Field(..., description='The version of the agent. Example: "1.0.0"') + documentation_url: str | None = Field(default=None, description='A URL providing additional documentation about the agent.') + capabilities: AgentCapabilities = Field(..., description='A2A Capability set supported by the agent.') + security_schemes: dict[str, SecurityScheme] | None = Field(default=None, description='The security scheme details used for authenticating with this agent.') + security_requirements: list[SecurityRequirement] | None = Field(default=None, description='Security requirements for contacting the agent.') + default_input_modes: list[str] = Field(..., description='protolint:enable REPEATED_FIELD_NAMES_PLURALIZED The set of interaction modes that the agent supports across all skills. This can be overridden per skill. Defined as media types.') + default_output_modes: list[str] = Field(..., description='The media types supported as outputs from this agent.') + skills: list[AgentSkill] = Field(..., description='Skills represent the abilities of an agent. It is largely a descriptive concept but represents a more focused set of behaviors that the agent is likely to succeed at.') + signatures: list[AgentCardSignature] | None = Field(default=None, description='JSON Web Signatures computed for this `AgentCard`.') + icon_url: str | None = Field(default=None, description='Optional. A URL to an icon for the agent.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class AgentExtension(A2ABaseModel): + """A declaration of a protocol extension supported by an Agent.""" + + uri: str = Field(default='', description='The unique URI identifying the extension.') + description: str = Field(default='', description='A human-readable description of how this agent uses the extension.') + required: bool = Field(default=False, description='If true, the client must understand and comply with the extension\'s requirements.') + params: dict[str, Any] = Field(default=None, description='Optional. Extension-specific configuration parameters.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class AgentCardSignature(A2ABaseModel): + """AgentCardSignature represents a JWS signature of an AgentCard. This follows the JSON format of an RFC 7515 JSON Web Signature (JWS).""" + + protected: str = Field(..., description='(-- api-linter: core::0140::reserved-words=disabled aip.dev/not-precedent: Backwards compatibility --) Required. The protected JWS header for the signature. This is always a base64url-encoded JSON object.') + signature: str = Field(..., description='Required. The computed signature, base64url-encoded.') + header: dict[str, Any] = Field(default=None, description='The unprotected JWS header values.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class StringList(A2ABaseModel): + """protolint:disable REPEATED_FIELD_NAMES_PLURALIZED A list of strings.""" + + list_: list[str] | None = Field(default=None, alias='list', description='The individual string values.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class SecurityRequirement(A2ABaseModel): + """Defines the security requirements for an agent.""" + + schemes: dict[str, StringList] | None = Field(default=None, description='A map of security schemes to the required scopes.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class SecurityScheme(A2ABaseModel): + """Defines a security scheme that can be used to secure an agent's endpoints. This is a discriminated union type based on the OpenAPI 3.2 Security Scheme Object. See: https://spec.openapis.org/oas/v3.2.0.html#security-scheme-object""" + + scheme: APIKeySecurityScheme | HTTPAuthSecurityScheme | OAuth2SecurityScheme | OpenIdConnectSecurityScheme | MutualTlsSecurityScheme | None = None + + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class APIKeySecurityScheme(A2ABaseModel): + """Defines a security scheme using an API key.""" + + description: str = Field(default='', description='An optional description for the security scheme.') + location: str = Field(..., description='The location of the API key. Valid values are "query", "header", or "cookie".') + name: str = Field(..., description='The name of the header, query, or cookie parameter to be used.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class HTTPAuthSecurityScheme(A2ABaseModel): + """Defines a security scheme using HTTP authentication.""" + + description: str = Field(default='', description='An optional description for the security scheme.') + scheme: str = Field(..., description='The name of the HTTP Authentication scheme to be used in the Authorization header, as defined in RFC7235 (e.g., "Bearer"). This value should be registered in the IANA Authentication Scheme registry.') + bearer_format: str = Field(default='', description='A hint to the client to identify how the bearer token is formatted (e.g., "JWT"). Primarily for documentation purposes.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class OAuthFlows(A2ABaseModel): + """Defines the configuration for the supported OAuth 2.0 flows.""" + + flow: AuthorizationCodeOAuthFlow | ClientCredentialsOAuthFlow | ImplicitOAuthFlow | PasswordOAuthFlow | DeviceCodeOAuthFlow | None = None + + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class OAuth2SecurityScheme(A2ABaseModel): + """Defines a security scheme using OAuth 2.0.""" + + description: str = Field(default='', description='An optional description for the security scheme.') + flows: OAuthFlows = Field(..., description='An object containing configuration information for the supported OAuth 2.0 flows.') + oauth2_metadata_url: str = Field(default='', description='URL to the OAuth2 authorization server metadata [RFC 8414](https://datatracker.ietf.org/doc/html/rfc8414). TLS is required.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class OpenIdConnectSecurityScheme(A2ABaseModel): + """Defines a security scheme using OpenID Connect.""" + + description: str = Field(default='', description='An optional description for the security scheme.') + open_id_connect_url: str = Field(..., description='The [OpenID Connect Discovery URL](https://openid.net/specs/openid-connect-discovery-1_0.html) for the OIDC provider\'s metadata.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class MutualTlsSecurityScheme(A2ABaseModel): + """Defines a security scheme using mTLS authentication.""" + + description: str = Field(default='', description='An optional description for the security scheme.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class AuthorizationCodeOAuthFlow(A2ABaseModel): + """Defines configuration details for the OAuth 2.0 Authorization Code flow.""" + + authorization_url: str = Field(..., description='The authorization URL to be used for this flow.') + token_url: str = Field(..., description='The token URL to be used for this flow.') + refresh_url: str = Field(default='', description='The URL to be used for obtaining refresh tokens.') + scopes: dict[str, str] = Field(..., description='The available scopes for the OAuth2 security scheme.') + pkce_required: bool = Field(default=False, description='Indicates if PKCE (RFC 7636) is required for this flow. PKCE should always be used for public clients and is recommended for all clients.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class ClientCredentialsOAuthFlow(A2ABaseModel): + """Defines configuration details for the OAuth 2.0 Client Credentials flow.""" + + token_url: str = Field(..., description='The token URL to be used for this flow.') + refresh_url: str = Field(default='', description='The URL to be used for obtaining refresh tokens.') + scopes: dict[str, str] = Field(..., description='The available scopes for the OAuth2 security scheme.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class ImplicitOAuthFlow(A2ABaseModel): + """Deprecated: Use Authorization Code + PKCE instead.""" + + authorization_url: str = Field(default='', description='The authorization URL to be used for this flow. This MUST be in the form of a URL. The OAuth2 standard requires the use of TLS') + refresh_url: str = Field(default='', description='The URL to be used for obtaining refresh tokens. This MUST be in the form of a URL. The OAuth2 standard requires the use of TLS.') + scopes: dict[str, str] | None = Field(default=None, description='The available scopes for the OAuth2 security scheme. A map between the scope name and a short description for it. The map MAY be empty.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class PasswordOAuthFlow(A2ABaseModel): + """Deprecated: Use Authorization Code + PKCE or Device Code.""" + + token_url: str = Field(default='', description='The token URL to be used for this flow. This MUST be in the form of a URL. The OAuth2 standard requires the use of TLS.') + refresh_url: str = Field(default='', description='The URL to be used for obtaining refresh tokens. This MUST be in the form of a URL. The OAuth2 standard requires the use of TLS.') + scopes: dict[str, str] | None = Field(default=None, description='The available scopes for the OAuth2 security scheme. A map between the scope name and a short description for it. The map MAY be empty.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class DeviceCodeOAuthFlow(A2ABaseModel): + """Defines configuration details for the OAuth 2.0 Device Code flow (RFC 8628). This flow is designed for input-constrained devices such as IoT devices, and CLI tools where the user authenticates on a separate device.""" + + device_authorization_url: str = Field(..., description='The device authorization endpoint URL.') + token_url: str = Field(..., description='The token URL to be used for this flow.') + refresh_url: str = Field(default='', description='The URL to be used for obtaining refresh tokens.') + scopes: dict[str, str] = Field(..., description='The available scopes for the OAuth2 security scheme.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class SendMessageRequest(A2ABaseModel): + """Represents a request for the `SendMessage` method.""" + + tenant: str = Field(default='', description='Optional. Tenant ID, provided as a path parameter.') + message: Message = Field(..., description='The message to send to the agent.') + configuration: SendMessageConfiguration = Field(default=None, description='Configuration for the send request.') + metadata: dict[str, Any] = Field(default=None, description='A flexible key-value map for passing additional context or parameters.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class GetTaskRequest(A2ABaseModel): + """Represents a request for the `GetTask` method.""" + + tenant: str = Field(default='', description='Optional. Tenant ID, provided as a path parameter.') + id: str = Field(..., description='The resource ID of the task to retrieve.') + history_length: int | None = Field(default=None, description='The maximum number of most recent messages from the task\'s history to retrieve. An unset value means the client does not impose any limit. A value of zero is a request to not include any messages. The server MUST NOT return more messages than the provided value, but MAY apply a lower limit.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class ListTasksRequest(A2ABaseModel): + """Parameters for listing tasks with optional filtering criteria.""" + + tenant: str = Field(default='', description='Tenant ID, provided as a path parameter.') + context_id: str = Field(default='', description='Filter tasks by context ID to get tasks from a specific conversation or session.') + status: TaskState = Field(default=None, description='Filter tasks by their current status state.') + page_size: int | None = Field(default=None, description='The maximum number of tasks to return. The service may return fewer than this value. If unspecified, at most 50 tasks will be returned. The minimum value is 1. The maximum value is 100.') + page_token: str = Field(default='', description='A page token, received from a previous `ListTasks` call. `ListTasksResponse.next_page_token`. Provide this to retrieve the subsequent page.') + history_length: int | None = Field(default=None, description='The maximum number of messages to include in each task\'s history.') + status_timestamp_after: datetime = Field(default=None, description='Filter tasks which have a status updated after the provided timestamp in ISO 8601 format (e.g., "2023-10-27T10:00:00Z"). Only tasks with a status timestamp time greater than or equal to this value will be returned.') + include_artifacts: bool | None = Field(default=None, description='Whether to include artifacts in the returned tasks. Defaults to false to reduce payload size.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + @field_serializer('status_timestamp_after') + @classmethod + def _serialize_timestamp(cls, v: datetime | None, _info: Any) -> str | None: + """Serialize datetime to RFC 3339 with UTC 'Z' suffix for ProtoJSON.""" + if v is None: + return None + return v.strftime('%Y-%m-%dT%H:%M:%S.') + f'{v.microsecond:06d}'[:3] + 'Z' + + + +class ListTasksResponse(A2ABaseModel): + """Result object for `ListTasks` method containing an array of tasks and pagination information.""" + + tasks: list[Task] = Field(..., description='Array of tasks matching the specified criteria.') + next_page_token: str = Field(..., description='A token to retrieve the next page of results, or empty if there are no more results in the list.') + page_size: int = Field(..., description='The page size used for this response.') + total_size: int = Field(..., description='Total number of tasks available (before pagination).') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class CancelTaskRequest(A2ABaseModel): + """Represents a request for the `CancelTask` method.""" + + tenant: str = Field(default='', description='Optional. Tenant ID, provided as a path parameter.') + id: str = Field(..., description='The resource ID of the task to cancel.') + metadata: dict[str, Any] = Field(default=None, description='A flexible key-value map for passing additional context or parameters.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class GetTaskPushNotificationConfigRequest(A2ABaseModel): + """Represents a request for the `GetTaskPushNotificationConfig` method.""" + + tenant: str = Field(default='', description='Optional. Tenant ID, provided as a path parameter.') + task_id: str = Field(..., description='The parent task resource ID.') + id: str = Field(..., description='The resource ID of the configuration to retrieve.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class DeleteTaskPushNotificationConfigRequest(A2ABaseModel): + """Represents a request for the `DeleteTaskPushNotificationConfig` method.""" + + tenant: str = Field(default='', description='Optional. Tenant ID, provided as a path parameter.') + task_id: str = Field(..., description='The parent task resource ID.') + id: str = Field(..., description='The resource ID of the configuration to delete.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class SubscribeToTaskRequest(A2ABaseModel): + """Represents a request for the `SubscribeToTask` method.""" + + tenant: str = Field(default='', description='Optional. Tenant ID, provided as a path parameter.') + id: str = Field(..., description='The resource ID of the task to subscribe to.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class ListTaskPushNotificationConfigsRequest(A2ABaseModel): + """Represents a request for the `ListTaskPushNotificationConfigs` method.""" + + tenant: str = Field(default='', description='Optional. Tenant ID, provided as a path parameter.') + task_id: str = Field(..., description='The parent task resource ID.') + page_size: int = Field(default=0, description='The maximum number of configurations to return.') + page_token: str = Field(default='', description='A page token received from a previous `ListTaskPushNotificationConfigsRequest` call.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class GetExtendedAgentCardRequest(A2ABaseModel): + """Represents a request for the `GetExtendedAgentCard` method.""" + + tenant: str = Field(default='', description='Optional. Tenant ID, provided as a path parameter.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class SendMessageResponse(A2ABaseModel): + """Represents the response for the `SendMessage` method.""" + + payload: Task | Message | None = None + + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class StreamResponse(A2ABaseModel): + """A wrapper object used in streaming operations to encapsulate different types of response data.""" + + payload: Task | Message | TaskStatusUpdateEvent | TaskArtifactUpdateEvent | None = None + + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + +class ListTaskPushNotificationConfigsResponse(A2ABaseModel): + """Represents a successful response for the `ListTaskPushNotificationConfigs` method.""" + + configs: list[TaskPushNotificationConfig] | None = Field(default=None, description='The list of push notification configurations.') + next_page_token: str = Field(default='', description='A token to retrieve the next page of results, or empty if there are no more results in the list.') + + def to_proto_json(self) -> dict: + """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" + return self.model_dump(by_alias=True, exclude_none=True) + + + + +__all__ = [ + "TaskState", + "Role", + "AuthenticationInfo", + "TaskPushNotificationConfig", + "SendMessageConfiguration", + "Part", + "Message", + "TaskStatus", + "Task", + "Artifact", + "TaskStatusUpdateEvent", + "TaskArtifactUpdateEvent", + "AgentInterface", + "AgentProvider", + "AgentCapabilities", + "AgentSkill", + "AgentCard", + "AgentExtension", + "AgentCardSignature", + "StringList", + "SecurityRequirement", + "SecurityScheme", + "APIKeySecurityScheme", + "HTTPAuthSecurityScheme", + "OAuthFlows", + "OAuth2SecurityScheme", + "OpenIdConnectSecurityScheme", + "MutualTlsSecurityScheme", + "AuthorizationCodeOAuthFlow", + "ClientCredentialsOAuthFlow", + "ImplicitOAuthFlow", + "PasswordOAuthFlow", + "DeviceCodeOAuthFlow", + "SendMessageRequest", + "GetTaskRequest", + "ListTasksRequest", + "ListTasksResponse", + "CancelTaskRequest", + "GetTaskPushNotificationConfigRequest", + "DeleteTaskPushNotificationConfigRequest", + "SubscribeToTaskRequest", + "ListTaskPushNotificationConfigsRequest", + "GetExtendedAgentCardRequest", + "SendMessageResponse", + "StreamResponse", + "ListTaskPushNotificationConfigsResponse", +] From 45904ac29a259280835d2bbb960be32557e1af18 Mon Sep 17 00:00:00 2001 From: brucearctor <5032356+brucearctor@users.noreply.github.com> Date: Thu, 26 Mar 2026 20:37:36 -0700 Subject: [PATCH 2/7] fix: regenerate a2a_pydantic.py with ruff-compliant output Generator improvements applied: - Single quotes in __all__ entries - Alphabetically sorted __all__ - Docstring trailing periods (PEP 257) - Deduplicated union types in oneof fields - Smart description quoting (double quotes when containing apostrophes) - isort-compliant import ordering (stdlib, third-party, local) - TYPE_CHECKING for datetime import with model_rebuild() - TYPE_CHECKING before Any in typing imports --- src/a2a/types/a2a_pydantic.py | 136 ++++++++++++++++++---------------- 1 file changed, 72 insertions(+), 64 deletions(-) diff --git a/src/a2a/types/a2a_pydantic.py b/src/a2a/types/a2a_pydantic.py index b1165ccbb..dad9c9433 100644 --- a/src/a2a/types/a2a_pydantic.py +++ b/src/a2a/types/a2a_pydantic.py @@ -2,14 +2,17 @@ from __future__ import annotations -from datetime import datetime from enum import Enum +from typing import TYPE_CHECKING, Any -from typing import Any -from a2a._base import A2ABaseModel +if TYPE_CHECKING: + from datetime import datetime + from pydantic import Field, field_serializer +from a2a._base import A2ABaseModel + class TaskState(str, Enum): """Defines the possible lifecycle states of a `Task`.""" @@ -66,7 +69,7 @@ class SendMessageConfiguration(A2ABaseModel): accepted_output_modes: list[str] | None = Field(default=None, description='A list of media types the client is prepared to accept for response parts. Agents SHOULD use this to tailor their output.') task_push_notification_config: TaskPushNotificationConfig = Field(default=None, description='Configuration for the agent to send push notifications for task updates. Task id should be empty when sending this configuration in a `SendMessage` request.') - history_length: int | None = Field(default=None, description='The maximum number of most recent messages from the task\'s history to retrieve in the response. An unset value means the client does not impose any limit. A value of zero is a request to not include any messages. The server MUST NOT return more messages than the provided value, but MAY apply a lower limit.') + history_length: int | None = Field(default=None, description="The maximum number of most recent messages from the task's history to retrieve in the response. An unset value means the client does not impose any limit. A value of zero is a request to not include any messages. The server MUST NOT return more messages than the provided value, but MAY apply a lower limit.") return_immediately: bool = Field(default=False, description='If `true`, the operation returns immediately after creating the task, even if processing is still in progress. If `false` (default), the operation MUST wait until the task reaches a terminal (`COMPLETED`, `FAILED`, `CANCELED`, `REJECTED`) or interrupted (`INPUT_REQUIRED`, `AUTH_REQUIRED`) state before returning.') def to_proto_json(self) -> dict: @@ -81,7 +84,7 @@ class Part(A2ABaseModel): metadata: dict[str, Any] = Field(default=None, description='Optional. metadata associated with this part.') filename: str = Field(default='', description='An optional `filename` for the file (e.g., "document.pdf").') media_type: str = Field(default='', description='The `media_type` (MIME type) of the part content (e.g., "text/plain", "application/json", "image/png"). This field is available for all part types.') - content: str | bytes | str | Any | None = None + content: str | bytes | Any | None = None def to_proto_json(self) -> dict: @@ -109,7 +112,7 @@ def to_proto_json(self) -> dict: class TaskStatus(A2ABaseModel): - """A container for the status of a task""" + """A container for the status of a task.""" state: TaskState = Field(..., description='The current state of this task.') message: Message = Field(default=None, description='A message associated with the status.') @@ -208,8 +211,8 @@ def to_proto_json(self) -> dict: class AgentProvider(A2ABaseModel): """Represents the service provider of an agent.""" - url: str = Field(..., description='A URL for the agent provider\'s website or relevant documentation. Example: "https://ai.google.dev"') - organization: str = Field(..., description='The name of the agent provider\'s organization. Example: "Google"') + url: str = Field(..., description="A URL for the agent provider's website or relevant documentation. Example: \"https://ai.google.dev\"") + organization: str = Field(..., description="The name of the agent provider's organization. Example: \"Google\"") def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" @@ -234,13 +237,13 @@ def to_proto_json(self) -> dict: class AgentSkill(A2ABaseModel): """Represents a distinct capability or function that an agent can perform.""" - id: str = Field(..., description='A unique identifier for the agent\'s skill.') + id: str = Field(..., description="A unique identifier for the agent's skill.") name: str = Field(..., description='A human-readable name for the skill.') description: str = Field(..., description='A detailed description of the skill.') - tags: list[str] = Field(..., description='A set of keywords describing the skill\'s capabilities.') + tags: list[str] = Field(..., description="A set of keywords describing the skill's capabilities.") examples: list[str] | None = Field(default=None, description='Example prompts or scenarios that this skill can handle.') - input_modes: list[str] | None = Field(default=None, description='The set of supported input media types for this skill, overriding the agent\'s defaults.') - output_modes: list[str] | None = Field(default=None, description='The set of supported output media types for this skill, overriding the agent\'s defaults.') + input_modes: list[str] | None = Field(default=None, description="The set of supported input media types for this skill, overriding the agent's defaults.") + output_modes: list[str] | None = Field(default=None, description="The set of supported output media types for this skill, overriding the agent's defaults.") security_requirements: list[SecurityRequirement] | None = Field(default=None, description='Security schemes necessary for this skill.') def to_proto_json(self) -> dict: @@ -250,7 +253,7 @@ def to_proto_json(self) -> dict: class AgentCard(A2ABaseModel): - """A self-describing manifest for an agent. It provides essential metadata including the agent's identity, capabilities, skills, supported communication methods, and security requirements. Next ID: 20""" + """A self-describing manifest for an agent. It provides essential metadata including the agent's identity, capabilities, skills, supported communication methods, and security requirements. Next ID: 20.""" name: str = Field(..., description='A human readable name for the agent. Example: "Recipe Agent"') description: str = Field(..., description='A human-readable description of the agent, assisting users and other agents in understanding its purpose. Example: "Agent that helps users with recipes and cooking."') @@ -278,7 +281,7 @@ class AgentExtension(A2ABaseModel): uri: str = Field(default='', description='The unique URI identifying the extension.') description: str = Field(default='', description='A human-readable description of how this agent uses the extension.') - required: bool = Field(default=False, description='If true, the client must understand and comply with the extension\'s requirements.') + required: bool = Field(default=False, description="If true, the client must understand and comply with the extension's requirements.") params: dict[str, Any] = Field(default=None, description='Optional. Extension-specific configuration parameters.') def to_proto_json(self) -> dict: @@ -323,7 +326,7 @@ def to_proto_json(self) -> dict: class SecurityScheme(A2ABaseModel): - """Defines a security scheme that can be used to secure an agent's endpoints. This is a discriminated union type based on the OpenAPI 3.2 Security Scheme Object. See: https://spec.openapis.org/oas/v3.2.0.html#security-scheme-object""" + """Defines a security scheme that can be used to secure an agent's endpoints. This is a discriminated union type based on the OpenAPI 3.2 Security Scheme Object. See: https://spec.openapis.org/oas/v3.2.0.html#security-scheme-object.""" scheme: APIKeySecurityScheme | HTTPAuthSecurityScheme | OAuth2SecurityScheme | OpenIdConnectSecurityScheme | MutualTlsSecurityScheme | None = None @@ -389,7 +392,7 @@ class OpenIdConnectSecurityScheme(A2ABaseModel): """Defines a security scheme using OpenID Connect.""" description: str = Field(default='', description='An optional description for the security scheme.') - open_id_connect_url: str = Field(..., description='The [OpenID Connect Discovery URL](https://openid.net/specs/openid-connect-discovery-1_0.html) for the OIDC provider\'s metadata.') + open_id_connect_url: str = Field(..., description="The [OpenID Connect Discovery URL](https://openid.net/specs/openid-connect-discovery-1_0.html) for the OIDC provider's metadata.") def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" @@ -495,7 +498,7 @@ class GetTaskRequest(A2ABaseModel): tenant: str = Field(default='', description='Optional. Tenant ID, provided as a path parameter.') id: str = Field(..., description='The resource ID of the task to retrieve.') - history_length: int | None = Field(default=None, description='The maximum number of most recent messages from the task\'s history to retrieve. An unset value means the client does not impose any limit. A value of zero is a request to not include any messages. The server MUST NOT return more messages than the provided value, but MAY apply a lower limit.') + history_length: int | None = Field(default=None, description="The maximum number of most recent messages from the task's history to retrieve. An unset value means the client does not impose any limit. A value of zero is a request to not include any messages. The server MUST NOT return more messages than the provided value, but MAY apply a lower limit.") def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" @@ -511,7 +514,7 @@ class ListTasksRequest(A2ABaseModel): status: TaskState = Field(default=None, description='Filter tasks by their current status state.') page_size: int | None = Field(default=None, description='The maximum number of tasks to return. The service may return fewer than this value. If unspecified, at most 50 tasks will be returned. The minimum value is 1. The maximum value is 100.') page_token: str = Field(default='', description='A page token, received from a previous `ListTasks` call. `ListTasksResponse.next_page_token`. Provide this to retrieve the subsequent page.') - history_length: int | None = Field(default=None, description='The maximum number of messages to include in each task\'s history.') + history_length: int | None = Field(default=None, description="The maximum number of messages to include in each task's history.") status_timestamp_after: datetime = Field(default=None, description='Filter tasks which have a status updated after the provided timestamp in ISO 8601 format (e.g., "2023-10-27T10:00:00Z"). Only tasks with a status timestamp time greater than or equal to this value will be returned.') include_artifacts: bool | None = Field(default=None, description='Whether to include artifacts in the returned tasks. Defaults to false to reduce payload size.') @@ -657,50 +660,55 @@ def to_proto_json(self) -> dict: __all__ = [ - "TaskState", - "Role", - "AuthenticationInfo", - "TaskPushNotificationConfig", - "SendMessageConfiguration", - "Part", - "Message", - "TaskStatus", - "Task", - "Artifact", - "TaskStatusUpdateEvent", - "TaskArtifactUpdateEvent", - "AgentInterface", - "AgentProvider", - "AgentCapabilities", - "AgentSkill", - "AgentCard", - "AgentExtension", - "AgentCardSignature", - "StringList", - "SecurityRequirement", - "SecurityScheme", - "APIKeySecurityScheme", - "HTTPAuthSecurityScheme", - "OAuthFlows", - "OAuth2SecurityScheme", - "OpenIdConnectSecurityScheme", - "MutualTlsSecurityScheme", - "AuthorizationCodeOAuthFlow", - "ClientCredentialsOAuthFlow", - "ImplicitOAuthFlow", - "PasswordOAuthFlow", - "DeviceCodeOAuthFlow", - "SendMessageRequest", - "GetTaskRequest", - "ListTasksRequest", - "ListTasksResponse", - "CancelTaskRequest", - "GetTaskPushNotificationConfigRequest", - "DeleteTaskPushNotificationConfigRequest", - "SubscribeToTaskRequest", - "ListTaskPushNotificationConfigsRequest", - "GetExtendedAgentCardRequest", - "SendMessageResponse", - "StreamResponse", - "ListTaskPushNotificationConfigsResponse", + 'APIKeySecurityScheme', + 'AgentCapabilities', + 'AgentCard', + 'AgentCardSignature', + 'AgentExtension', + 'AgentInterface', + 'AgentProvider', + 'AgentSkill', + 'Artifact', + 'AuthenticationInfo', + 'AuthorizationCodeOAuthFlow', + 'CancelTaskRequest', + 'ClientCredentialsOAuthFlow', + 'DeleteTaskPushNotificationConfigRequest', + 'DeviceCodeOAuthFlow', + 'GetExtendedAgentCardRequest', + 'GetTaskPushNotificationConfigRequest', + 'GetTaskRequest', + 'HTTPAuthSecurityScheme', + 'ImplicitOAuthFlow', + 'ListTaskPushNotificationConfigsRequest', + 'ListTaskPushNotificationConfigsResponse', + 'ListTasksRequest', + 'ListTasksResponse', + 'Message', + 'MutualTlsSecurityScheme', + 'OAuth2SecurityScheme', + 'OAuthFlows', + 'OpenIdConnectSecurityScheme', + 'Part', + 'PasswordOAuthFlow', + 'Role', + 'SecurityRequirement', + 'SecurityScheme', + 'SendMessageConfiguration', + 'SendMessageRequest', + 'SendMessageResponse', + 'StreamResponse', + 'StringList', + 'SubscribeToTaskRequest', + 'Task', + 'TaskArtifactUpdateEvent', + 'TaskPushNotificationConfig', + 'TaskState', + 'TaskStatus', + 'TaskStatusUpdateEvent', ] + + +# Rebuild models that use TYPE_CHECKING imports (forward references) +TaskStatus.model_rebuild() +ListTasksRequest.model_rebuild() From b933810551e1cdbff8ed0d55ada6c42f67b47509 Mon Sep 17 00:00:00 2001 From: brucearctor <5032356+brucearctor@users.noreply.github.com> Date: Thu, 26 Mar 2026 20:50:53 -0700 Subject: [PATCH 3/7] fix: update proto2pydantic to v0.5.0 with ruff-compliant output - gen_proto.sh: bump proto2pydantic v0.4.0 -> v0.5.0 - a2a_pydantic.py: improved millisecond formatting in timestamp serializer --- scripts/gen_proto.sh | 4 ++-- src/a2a/types/a2a_pydantic.py | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/scripts/gen_proto.sh b/scripts/gen_proto.sh index 828f5322a..f355254b8 100755 --- a/scripts/gen_proto.sh +++ b/scripts/gen_proto.sh @@ -2,8 +2,8 @@ set -e # Install proto2pydantic plugin for Pydantic model generation -echo "Installing protoc-gen-proto2pydantic@v0.4.0..." -go install github.com/protocgen/proto2pydantic@v0.4.0 +echo "Installing protoc-gen-proto2pydantic@v0.5.0..." +go install github.com/protocgen/proto2pydantic@v0.5.0 # Run buf generate to regenerate protobuf code and OpenAPI spec npx --yes @bufbuild/buf generate diff --git a/src/a2a/types/a2a_pydantic.py b/src/a2a/types/a2a_pydantic.py index dad9c9433..0a694d80c 100644 --- a/src/a2a/types/a2a_pydantic.py +++ b/src/a2a/types/a2a_pydantic.py @@ -128,7 +128,7 @@ def _serialize_timestamp(cls, v: datetime | None, _info: Any) -> str | None: """Serialize datetime to RFC 3339 with UTC 'Z' suffix for ProtoJSON.""" if v is None: return None - return v.strftime('%Y-%m-%dT%H:%M:%S.') + f'{v.microsecond:06d}'[:3] + 'Z' + return v.strftime('%Y-%m-%dT%H:%M:%S.') + f'{v.microsecond // 1000:03d}' + 'Z' @@ -528,7 +528,7 @@ def _serialize_timestamp(cls, v: datetime | None, _info: Any) -> str | None: """Serialize datetime to RFC 3339 with UTC 'Z' suffix for ProtoJSON.""" if v is None: return None - return v.strftime('%Y-%m-%dT%H:%M:%S.') + f'{v.microsecond:06d}'[:3] + 'Z' + return v.strftime('%Y-%m-%dT%H:%M:%S.') + f'{v.microsecond // 1000:03d}' + 'Z' From e7111b079fb51bcccf1e2bf65b78f53267f47283 Mon Sep 17 00:00:00 2001 From: brucearctor <5032356+brucearctor@users.noreply.github.com> Date: Thu, 26 Mar 2026 21:01:35 -0700 Subject: [PATCH 4/7] fix: bump proto2pydantic to v0.5.1 --- scripts/gen_proto.sh | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/scripts/gen_proto.sh b/scripts/gen_proto.sh index f355254b8..e035b92e3 100755 --- a/scripts/gen_proto.sh +++ b/scripts/gen_proto.sh @@ -2,8 +2,8 @@ set -e # Install proto2pydantic plugin for Pydantic model generation -echo "Installing protoc-gen-proto2pydantic@v0.5.0..." -go install github.com/protocgen/proto2pydantic@v0.5.0 +echo "Installing protoc-gen-proto2pydantic@v0.5.1..." +go install github.com/protocgen/proto2pydantic@v0.5.1 # Run buf generate to regenerate protobuf code and OpenAPI spec npx --yes @bufbuild/buf generate From 226d4068616042b98c442412637028a40937663b Mon Sep 17 00:00:00 2001 From: brucearctor <5032356+brucearctor@users.noreply.github.com> Date: Thu, 26 Mar 2026 21:12:32 -0700 Subject: [PATCH 5/7] fix: add spelling allowlist entry and jscpd exclusion for generated code - Add 'protocgen' to spelling allow.txt (proto2pydantic plugin org) - Exclude generated *_pydantic.py from jscpd copy/paste detection --- .github/actions/spelling/allow.txt | 1 + .jscpd.json | 3 ++- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/.github/actions/spelling/allow.txt b/.github/actions/spelling/allow.txt index df74a242d..5e5c61f44 100644 --- a/.github/actions/spelling/allow.txt +++ b/.github/actions/spelling/allow.txt @@ -107,6 +107,7 @@ proto protobuf Protobuf protoc +protocgen pydantic pyi pypistats diff --git a/.jscpd.json b/.jscpd.json index ed59a6491..3d15e2edd 100644 --- a/.jscpd.json +++ b/.jscpd.json @@ -6,7 +6,8 @@ "**/src/a2a/grpc/**", "**/src/a2a/compat/**", "**/.nox/**", - "**/.venv/**" + "**/.venv/**", + "**/src/a2a/types/*_pydantic.py" ], "threshold": 3, "reporters": ["html", "markdown"] From fb8a0faaf4d44ae600476004cd92d440723f1670 Mon Sep 17 00:00:00 2001 From: brucearctor <5032356+brucearctor@users.noreply.github.com> Date: Thu, 26 Mar 2026 21:44:32 -0700 Subject: [PATCH 6/7] fix: update proto2pydantic to v0.5.2 with proper type annotations - Enum fields default to zero value (e.g., TaskState.TASK_STATE_UNSPECIFIED) - Message fields properly annotated as | None - Fixes all mypy and pyright type errors in generated code --- scripts/gen_proto.sh | 4 ++-- src/a2a/types/a2a_pydantic.py | 36 +++++++++++++++++------------------ 2 files changed, 20 insertions(+), 20 deletions(-) diff --git a/scripts/gen_proto.sh b/scripts/gen_proto.sh index e035b92e3..37f8c2c0f 100755 --- a/scripts/gen_proto.sh +++ b/scripts/gen_proto.sh @@ -2,8 +2,8 @@ set -e # Install proto2pydantic plugin for Pydantic model generation -echo "Installing protoc-gen-proto2pydantic@v0.5.1..." -go install github.com/protocgen/proto2pydantic@v0.5.1 +echo "Installing protoc-gen-proto2pydantic@v0.5.2..." +go install github.com/protocgen/proto2pydantic@v0.5.2 # Run buf generate to regenerate protobuf code and OpenAPI spec npx --yes @bufbuild/buf generate diff --git a/src/a2a/types/a2a_pydantic.py b/src/a2a/types/a2a_pydantic.py index 0a694d80c..d80fc326f 100644 --- a/src/a2a/types/a2a_pydantic.py +++ b/src/a2a/types/a2a_pydantic.py @@ -56,7 +56,7 @@ class TaskPushNotificationConfig(A2ABaseModel): task_id: str = Field(default='', description='The ID of the task this configuration is associated with.') url: str = Field(..., description='The URL where the notification should be sent.') token: str = Field(default='', description='A token unique for this task or session.') - authentication: AuthenticationInfo = Field(default=None, description='Authentication information required to send the notification.') + authentication: AuthenticationInfo | None = Field(default=None, description='Authentication information required to send the notification.') def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" @@ -68,7 +68,7 @@ class SendMessageConfiguration(A2ABaseModel): """Configuration of a send message request.""" accepted_output_modes: list[str] | None = Field(default=None, description='A list of media types the client is prepared to accept for response parts. Agents SHOULD use this to tailor their output.') - task_push_notification_config: TaskPushNotificationConfig = Field(default=None, description='Configuration for the agent to send push notifications for task updates. Task id should be empty when sending this configuration in a `SendMessage` request.') + task_push_notification_config: TaskPushNotificationConfig | None = Field(default=None, description='Configuration for the agent to send push notifications for task updates. Task id should be empty when sending this configuration in a `SendMessage` request.') history_length: int | None = Field(default=None, description="The maximum number of most recent messages from the task's history to retrieve in the response. An unset value means the client does not impose any limit. A value of zero is a request to not include any messages. The server MUST NOT return more messages than the provided value, but MAY apply a lower limit.") return_immediately: bool = Field(default=False, description='If `true`, the operation returns immediately after creating the task, even if processing is still in progress. If `false` (default), the operation MUST wait until the task reaches a terminal (`COMPLETED`, `FAILED`, `CANCELED`, `REJECTED`) or interrupted (`INPUT_REQUIRED`, `AUTH_REQUIRED`) state before returning.') @@ -81,7 +81,7 @@ def to_proto_json(self) -> dict: class Part(A2ABaseModel): """`Part` represents a container for a section of communication content. Parts can be purely textual, some sort of file (image, video, etc) or a structured data blob (i.e. JSON).""" - metadata: dict[str, Any] = Field(default=None, description='Optional. metadata associated with this part.') + metadata: dict[str, Any] | None = Field(default=None, description='Optional. metadata associated with this part.') filename: str = Field(default='', description='An optional `filename` for the file (e.g., "document.pdf").') media_type: str = Field(default='', description='The `media_type` (MIME type) of the part content (e.g., "text/plain", "application/json", "image/png"). This field is available for all part types.') content: str | bytes | Any | None = None @@ -101,7 +101,7 @@ class Message(A2ABaseModel): task_id: str = Field(default='', description='Optional. The task id of the message. If set, the message will be associated with the given task.') role: Role = Field(..., description='Identifies the sender of the message.') parts: list[Part] = Field(..., description='Parts is the container of the message content.') - metadata: dict[str, Any] = Field(default=None, description='Optional. Any metadata to provide along with the message.') + metadata: dict[str, Any] | None = Field(default=None, description='Optional. Any metadata to provide along with the message.') extensions: list[str] | None = Field(default=None, description='The URIs of extensions that are present or contributed to this Message.') reference_task_ids: list[str] | None = Field(default=None, description='A list of task IDs that this message references for additional context.') @@ -115,8 +115,8 @@ class TaskStatus(A2ABaseModel): """A container for the status of a task.""" state: TaskState = Field(..., description='The current state of this task.') - message: Message = Field(default=None, description='A message associated with the status.') - timestamp: datetime = Field(default=None, description='ISO 8601 Timestamp when the status was recorded. Example: "2023-10-27T10:00:00Z"') + message: Message | None = Field(default=None, description='A message associated with the status.') + timestamp: datetime | None = Field(default=None, description='ISO 8601 Timestamp when the status was recorded. Example: "2023-10-27T10:00:00Z"') def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" @@ -140,7 +140,7 @@ class Task(A2ABaseModel): status: TaskStatus = Field(..., description='The current status of a `Task`, including `state` and a `message`.') artifacts: list[Artifact] | None = Field(default=None, description='A set of output artifacts for a `Task`.') history: list[Message] | None = Field(default=None, description='protolint:disable REPEATED_FIELD_NAMES_PLURALIZED The history of interactions from a `Task`.') - metadata: dict[str, Any] = Field(default=None, description='protolint:enable REPEATED_FIELD_NAMES_PLURALIZED A key/value object to store custom metadata about a task.') + metadata: dict[str, Any] | None = Field(default=None, description='protolint:enable REPEATED_FIELD_NAMES_PLURALIZED A key/value object to store custom metadata about a task.') def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" @@ -155,7 +155,7 @@ class Artifact(A2ABaseModel): name: str = Field(default='', description='A human readable name for the artifact.') description: str = Field(default='', description='Optional. A human readable description of the artifact.') parts: list[Part] = Field(..., description='The content of the artifact. Must contain at least one part.') - metadata: dict[str, Any] = Field(default=None, description='Optional. Metadata included with the artifact.') + metadata: dict[str, Any] | None = Field(default=None, description='Optional. Metadata included with the artifact.') extensions: list[str] | None = Field(default=None, description='The URIs of extensions that are present or contributed to this Artifact.') def to_proto_json(self) -> dict: @@ -170,7 +170,7 @@ class TaskStatusUpdateEvent(A2ABaseModel): task_id: str = Field(..., description='The ID of the task that has changed.') context_id: str = Field(..., description='The ID of the context that the task belongs to.') status: TaskStatus = Field(..., description='The new status of the task.') - metadata: dict[str, Any] = Field(default=None, description='Optional. Metadata associated with the task update.') + metadata: dict[str, Any] | None = Field(default=None, description='Optional. Metadata associated with the task update.') def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" @@ -186,7 +186,7 @@ class TaskArtifactUpdateEvent(A2ABaseModel): artifact: Artifact = Field(..., description='The artifact that was generated or updated.') append: bool = Field(default=False, description='If true, the content of this artifact should be appended to a previously sent artifact with the same ID.') last_chunk: bool = Field(default=False, description='If true, this is the final chunk of the artifact.') - metadata: dict[str, Any] = Field(default=None, description='Optional. Metadata associated with the artifact update.') + metadata: dict[str, Any] | None = Field(default=None, description='Optional. Metadata associated with the artifact update.') def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" @@ -258,7 +258,7 @@ class AgentCard(A2ABaseModel): name: str = Field(..., description='A human readable name for the agent. Example: "Recipe Agent"') description: str = Field(..., description='A human-readable description of the agent, assisting users and other agents in understanding its purpose. Example: "Agent that helps users with recipes and cooking."') supported_interfaces: list[AgentInterface] = Field(..., description='Ordered list of supported interfaces. The first entry is preferred.') - provider: AgentProvider = Field(default=None, description='The service provider of the agent.') + provider: AgentProvider | None = Field(default=None, description='The service provider of the agent.') version: str = Field(..., description='The version of the agent. Example: "1.0.0"') documentation_url: str | None = Field(default=None, description='A URL providing additional documentation about the agent.') capabilities: AgentCapabilities = Field(..., description='A2A Capability set supported by the agent.') @@ -282,7 +282,7 @@ class AgentExtension(A2ABaseModel): uri: str = Field(default='', description='The unique URI identifying the extension.') description: str = Field(default='', description='A human-readable description of how this agent uses the extension.') required: bool = Field(default=False, description="If true, the client must understand and comply with the extension's requirements.") - params: dict[str, Any] = Field(default=None, description='Optional. Extension-specific configuration parameters.') + params: dict[str, Any] | None = Field(default=None, description='Optional. Extension-specific configuration parameters.') def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" @@ -295,7 +295,7 @@ class AgentCardSignature(A2ABaseModel): protected: str = Field(..., description='(-- api-linter: core::0140::reserved-words=disabled aip.dev/not-precedent: Backwards compatibility --) Required. The protected JWS header for the signature. This is always a base64url-encoded JSON object.') signature: str = Field(..., description='Required. The computed signature, base64url-encoded.') - header: dict[str, Any] = Field(default=None, description='The unprotected JWS header values.') + header: dict[str, Any] | None = Field(default=None, description='The unprotected JWS header values.') def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" @@ -484,8 +484,8 @@ class SendMessageRequest(A2ABaseModel): tenant: str = Field(default='', description='Optional. Tenant ID, provided as a path parameter.') message: Message = Field(..., description='The message to send to the agent.') - configuration: SendMessageConfiguration = Field(default=None, description='Configuration for the send request.') - metadata: dict[str, Any] = Field(default=None, description='A flexible key-value map for passing additional context or parameters.') + configuration: SendMessageConfiguration | None = Field(default=None, description='Configuration for the send request.') + metadata: dict[str, Any] | None = Field(default=None, description='A flexible key-value map for passing additional context or parameters.') def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" @@ -511,11 +511,11 @@ class ListTasksRequest(A2ABaseModel): tenant: str = Field(default='', description='Tenant ID, provided as a path parameter.') context_id: str = Field(default='', description='Filter tasks by context ID to get tasks from a specific conversation or session.') - status: TaskState = Field(default=None, description='Filter tasks by their current status state.') + status: TaskState = Field(default=TaskState.TASK_STATE_UNSPECIFIED, description='Filter tasks by their current status state.') page_size: int | None = Field(default=None, description='The maximum number of tasks to return. The service may return fewer than this value. If unspecified, at most 50 tasks will be returned. The minimum value is 1. The maximum value is 100.') page_token: str = Field(default='', description='A page token, received from a previous `ListTasks` call. `ListTasksResponse.next_page_token`. Provide this to retrieve the subsequent page.') history_length: int | None = Field(default=None, description="The maximum number of messages to include in each task's history.") - status_timestamp_after: datetime = Field(default=None, description='Filter tasks which have a status updated after the provided timestamp in ISO 8601 format (e.g., "2023-10-27T10:00:00Z"). Only tasks with a status timestamp time greater than or equal to this value will be returned.') + status_timestamp_after: datetime | None = Field(default=None, description='Filter tasks which have a status updated after the provided timestamp in ISO 8601 format (e.g., "2023-10-27T10:00:00Z"). Only tasks with a status timestamp time greater than or equal to this value will be returned.') include_artifacts: bool | None = Field(default=None, description='Whether to include artifacts in the returned tasks. Defaults to false to reduce payload size.') def to_proto_json(self) -> dict: @@ -551,7 +551,7 @@ class CancelTaskRequest(A2ABaseModel): tenant: str = Field(default='', description='Optional. Tenant ID, provided as a path parameter.') id: str = Field(..., description='The resource ID of the task to cancel.') - metadata: dict[str, Any] = Field(default=None, description='A flexible key-value map for passing additional context or parameters.') + metadata: dict[str, Any] | None = Field(default=None, description='A flexible key-value map for passing additional context or parameters.') def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" From 29e61b37f2ad38d894dc1dad040942560d70d1c1 Mon Sep 17 00:00:00 2001 From: brucearctor <5032356+brucearctor@users.noreply.github.com> Date: Thu, 26 Mar 2026 21:59:10 -0700 Subject: [PATCH 7/7] fix: add ruff format post-generation step and apply formatting - Add 'ruff format' to gen_proto.sh after buf generate - Apply formatting to a2a_pydantic.py (line-length wrapping for long Field() descriptions, consistent blank lines) --- scripts/gen_proto.sh | 4 + src/a2a/types/a2a_pydantic.py | 773 +++++++++++++++++++++++++--------- 2 files changed, 576 insertions(+), 201 deletions(-) diff --git a/scripts/gen_proto.sh b/scripts/gen_proto.sh index 37f8c2c0f..036fb5e38 100755 --- a/scripts/gen_proto.sh +++ b/scripts/gen_proto.sh @@ -8,6 +8,10 @@ go install github.com/protocgen/proto2pydantic@v0.5.2 # Run buf generate to regenerate protobuf code and OpenAPI spec npx --yes @bufbuild/buf generate +# Format generated Pydantic models (line-length wrapping for long Field() descriptions) +echo "Formatting generated Pydantic models..." +ruff format src/a2a/types/a2a_pydantic.py + # The OpenAPI generator produces a file named like 'a2a.swagger.json' or similar. # We need it to be 'a2a.json' for the A2A SDK. # Find the generated json file in the output directory diff --git a/src/a2a/types/a2a_pydantic.py b/src/a2a/types/a2a_pydantic.py index d80fc326f..b2466324c 100644 --- a/src/a2a/types/a2a_pydantic.py +++ b/src/a2a/types/a2a_pydantic.py @@ -39,84 +39,143 @@ class Role(str, Enum): class AuthenticationInfo(A2ABaseModel): """Defines authentication details, used for push notifications.""" - scheme: str = Field(..., description='HTTP Authentication Scheme from the [IANA registry](https://www.iana.org/assignments/http-authschemes/). Examples: `Bearer`, `Basic`, `Digest`. Scheme names are case-insensitive per [RFC 9110 Section 11.1](https://www.rfc-editor.org/rfc/rfc9110#section-11.1).') - credentials: str = Field(default='', description='Push Notification credentials. Format depends on the scheme (e.g., token for Bearer).') + scheme: str = Field( + ..., + description='HTTP Authentication Scheme from the [IANA registry](https://www.iana.org/assignments/http-authschemes/). Examples: `Bearer`, `Basic`, `Digest`. Scheme names are case-insensitive per [RFC 9110 Section 11.1](https://www.rfc-editor.org/rfc/rfc9110#section-11.1).', + ) + credentials: str = Field( + default='', + description='Push Notification credentials. Format depends on the scheme (e.g., token for Bearer).', + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class TaskPushNotificationConfig(A2ABaseModel): """A container associating a push notification configuration with a specific task.""" tenant: str = Field(default='', description='Optional. Tenant ID.') - id: str = Field(default='', description='The push notification configuration details. A unique identifier (e.g. UUID) for this push notification configuration.') - task_id: str = Field(default='', description='The ID of the task this configuration is associated with.') - url: str = Field(..., description='The URL where the notification should be sent.') - token: str = Field(default='', description='A token unique for this task or session.') - authentication: AuthenticationInfo | None = Field(default=None, description='Authentication information required to send the notification.') + id: str = Field( + default='', + description='The push notification configuration details. A unique identifier (e.g. UUID) for this push notification configuration.', + ) + task_id: str = Field( + default='', + description='The ID of the task this configuration is associated with.', + ) + url: str = Field( + ..., description='The URL where the notification should be sent.' + ) + token: str = Field( + default='', description='A token unique for this task or session.' + ) + authentication: AuthenticationInfo | None = Field( + default=None, + description='Authentication information required to send the notification.', + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class SendMessageConfiguration(A2ABaseModel): """Configuration of a send message request.""" - accepted_output_modes: list[str] | None = Field(default=None, description='A list of media types the client is prepared to accept for response parts. Agents SHOULD use this to tailor their output.') - task_push_notification_config: TaskPushNotificationConfig | None = Field(default=None, description='Configuration for the agent to send push notifications for task updates. Task id should be empty when sending this configuration in a `SendMessage` request.') - history_length: int | None = Field(default=None, description="The maximum number of most recent messages from the task's history to retrieve in the response. An unset value means the client does not impose any limit. A value of zero is a request to not include any messages. The server MUST NOT return more messages than the provided value, but MAY apply a lower limit.") - return_immediately: bool = Field(default=False, description='If `true`, the operation returns immediately after creating the task, even if processing is still in progress. If `false` (default), the operation MUST wait until the task reaches a terminal (`COMPLETED`, `FAILED`, `CANCELED`, `REJECTED`) or interrupted (`INPUT_REQUIRED`, `AUTH_REQUIRED`) state before returning.') + accepted_output_modes: list[str] | None = Field( + default=None, + description='A list of media types the client is prepared to accept for response parts. Agents SHOULD use this to tailor their output.', + ) + task_push_notification_config: TaskPushNotificationConfig | None = Field( + default=None, + description='Configuration for the agent to send push notifications for task updates. Task id should be empty when sending this configuration in a `SendMessage` request.', + ) + history_length: int | None = Field( + default=None, + description="The maximum number of most recent messages from the task's history to retrieve in the response. An unset value means the client does not impose any limit. A value of zero is a request to not include any messages. The server MUST NOT return more messages than the provided value, but MAY apply a lower limit.", + ) + return_immediately: bool = Field( + default=False, + description='If `true`, the operation returns immediately after creating the task, even if processing is still in progress. If `false` (default), the operation MUST wait until the task reaches a terminal (`COMPLETED`, `FAILED`, `CANCELED`, `REJECTED`) or interrupted (`INPUT_REQUIRED`, `AUTH_REQUIRED`) state before returning.', + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class Part(A2ABaseModel): """`Part` represents a container for a section of communication content. Parts can be purely textual, some sort of file (image, video, etc) or a structured data blob (i.e. JSON).""" - metadata: dict[str, Any] | None = Field(default=None, description='Optional. metadata associated with this part.') - filename: str = Field(default='', description='An optional `filename` for the file (e.g., "document.pdf").') - media_type: str = Field(default='', description='The `media_type` (MIME type) of the part content (e.g., "text/plain", "application/json", "image/png"). This field is available for all part types.') + metadata: dict[str, Any] | None = Field( + default=None, + description='Optional. metadata associated with this part.', + ) + filename: str = Field( + default='', + description='An optional `filename` for the file (e.g., "document.pdf").', + ) + media_type: str = Field( + default='', + description='The `media_type` (MIME type) of the part content (e.g., "text/plain", "application/json", "image/png"). This field is available for all part types.', + ) content: str | bytes | Any | None = None - def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class Message(A2ABaseModel): """`Message` is one unit of communication between client and server. It can be associated with a context and/or a task. For server messages, `context_id` must be provided, and `task_id` only if a task was created. For client messages, both fields are optional, with the caveat that if both are provided, they have to match (the `context_id` has to be the one that is set on the task). If only `task_id` is provided, the server will infer `context_id` from it.""" - message_id: str = Field(..., description='The unique identifier (e.g. UUID) of the message. This is created by the message creator.') - context_id: str = Field(default='', description='Optional. The context id of the message. If set, the message will be associated with the given context.') - task_id: str = Field(default='', description='Optional. The task id of the message. If set, the message will be associated with the given task.') + message_id: str = Field( + ..., + description='The unique identifier (e.g. UUID) of the message. This is created by the message creator.', + ) + context_id: str = Field( + default='', + description='Optional. The context id of the message. If set, the message will be associated with the given context.', + ) + task_id: str = Field( + default='', + description='Optional. The task id of the message. If set, the message will be associated with the given task.', + ) role: Role = Field(..., description='Identifies the sender of the message.') - parts: list[Part] = Field(..., description='Parts is the container of the message content.') - metadata: dict[str, Any] | None = Field(default=None, description='Optional. Any metadata to provide along with the message.') - extensions: list[str] | None = Field(default=None, description='The URIs of extensions that are present or contributed to this Message.') - reference_task_ids: list[str] | None = Field(default=None, description='A list of task IDs that this message references for additional context.') + parts: list[Part] = Field( + ..., description='Parts is the container of the message content.' + ) + metadata: dict[str, Any] | None = Field( + default=None, + description='Optional. Any metadata to provide along with the message.', + ) + extensions: list[str] | None = Field( + default=None, + description='The URIs of extensions that are present or contributed to this Message.', + ) + reference_task_ids: list[str] | None = Field( + default=None, + description='A list of task IDs that this message references for additional context.', + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class TaskStatus(A2ABaseModel): """A container for the status of a task.""" state: TaskState = Field(..., description='The current state of this task.') - message: Message | None = Field(default=None, description='A message associated with the status.') - timestamp: datetime | None = Field(default=None, description='ISO 8601 Timestamp when the status was recorded. Example: "2023-10-27T10:00:00Z"') + message: Message | None = Field( + default=None, description='A message associated with the status.' + ) + timestamp: datetime | None = Field( + default=None, + description='ISO 8601 Timestamp when the status was recorded. Example: "2023-10-27T10:00:00Z"', + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" @@ -128,395 +187,668 @@ def _serialize_timestamp(cls, v: datetime | None, _info: Any) -> str | None: """Serialize datetime to RFC 3339 with UTC 'Z' suffix for ProtoJSON.""" if v is None: return None - return v.strftime('%Y-%m-%dT%H:%M:%S.') + f'{v.microsecond // 1000:03d}' + 'Z' - + return ( + v.strftime('%Y-%m-%dT%H:%M:%S.') + + f'{v.microsecond // 1000:03d}' + + 'Z' + ) class Task(A2ABaseModel): """`Task` is the core unit of action for A2A. It has a current status and when results are created for the task they are stored in the artifact. If there are multiple turns for a task, these are stored in history.""" - id: str = Field(..., description='Unique identifier (e.g. UUID) for the task, generated by the server for a new task.') - context_id: str = Field(default='', description='Unique identifier (e.g. UUID) for the contextual collection of interactions (tasks and messages).') - status: TaskStatus = Field(..., description='The current status of a `Task`, including `state` and a `message`.') - artifacts: list[Artifact] | None = Field(default=None, description='A set of output artifacts for a `Task`.') - history: list[Message] | None = Field(default=None, description='protolint:disable REPEATED_FIELD_NAMES_PLURALIZED The history of interactions from a `Task`.') - metadata: dict[str, Any] | None = Field(default=None, description='protolint:enable REPEATED_FIELD_NAMES_PLURALIZED A key/value object to store custom metadata about a task.') + id: str = Field( + ..., + description='Unique identifier (e.g. UUID) for the task, generated by the server for a new task.', + ) + context_id: str = Field( + default='', + description='Unique identifier (e.g. UUID) for the contextual collection of interactions (tasks and messages).', + ) + status: TaskStatus = Field( + ..., + description='The current status of a `Task`, including `state` and a `message`.', + ) + artifacts: list[Artifact] | None = Field( + default=None, description='A set of output artifacts for a `Task`.' + ) + history: list[Message] | None = Field( + default=None, + description='protolint:disable REPEATED_FIELD_NAMES_PLURALIZED The history of interactions from a `Task`.', + ) + metadata: dict[str, Any] | None = Field( + default=None, + description='protolint:enable REPEATED_FIELD_NAMES_PLURALIZED A key/value object to store custom metadata about a task.', + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class Artifact(A2ABaseModel): """Artifacts represent task outputs.""" - artifact_id: str = Field(..., description='Unique identifier (e.g. UUID) for the artifact. It must be unique within a task.') - name: str = Field(default='', description='A human readable name for the artifact.') - description: str = Field(default='', description='Optional. A human readable description of the artifact.') - parts: list[Part] = Field(..., description='The content of the artifact. Must contain at least one part.') - metadata: dict[str, Any] | None = Field(default=None, description='Optional. Metadata included with the artifact.') - extensions: list[str] | None = Field(default=None, description='The URIs of extensions that are present or contributed to this Artifact.') + artifact_id: str = Field( + ..., + description='Unique identifier (e.g. UUID) for the artifact. It must be unique within a task.', + ) + name: str = Field( + default='', description='A human readable name for the artifact.' + ) + description: str = Field( + default='', + description='Optional. A human readable description of the artifact.', + ) + parts: list[Part] = Field( + ..., + description='The content of the artifact. Must contain at least one part.', + ) + metadata: dict[str, Any] | None = Field( + default=None, + description='Optional. Metadata included with the artifact.', + ) + extensions: list[str] | None = Field( + default=None, + description='The URIs of extensions that are present or contributed to this Artifact.', + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class TaskStatusUpdateEvent(A2ABaseModel): """An event sent by the agent to notify the client of a change in a task's status.""" - task_id: str = Field(..., description='The ID of the task that has changed.') - context_id: str = Field(..., description='The ID of the context that the task belongs to.') + task_id: str = Field( + ..., description='The ID of the task that has changed.' + ) + context_id: str = Field( + ..., description='The ID of the context that the task belongs to.' + ) status: TaskStatus = Field(..., description='The new status of the task.') - metadata: dict[str, Any] | None = Field(default=None, description='Optional. Metadata associated with the task update.') + metadata: dict[str, Any] | None = Field( + default=None, + description='Optional. Metadata associated with the task update.', + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class TaskArtifactUpdateEvent(A2ABaseModel): """A task delta where an artifact has been generated.""" - task_id: str = Field(..., description='The ID of the task for this artifact.') - context_id: str = Field(..., description='The ID of the context that this task belongs to.') - artifact: Artifact = Field(..., description='The artifact that was generated or updated.') - append: bool = Field(default=False, description='If true, the content of this artifact should be appended to a previously sent artifact with the same ID.') - last_chunk: bool = Field(default=False, description='If true, this is the final chunk of the artifact.') - metadata: dict[str, Any] | None = Field(default=None, description='Optional. Metadata associated with the artifact update.') + task_id: str = Field( + ..., description='The ID of the task for this artifact.' + ) + context_id: str = Field( + ..., description='The ID of the context that this task belongs to.' + ) + artifact: Artifact = Field( + ..., description='The artifact that was generated or updated.' + ) + append: bool = Field( + default=False, + description='If true, the content of this artifact should be appended to a previously sent artifact with the same ID.', + ) + last_chunk: bool = Field( + default=False, + description='If true, this is the final chunk of the artifact.', + ) + metadata: dict[str, Any] | None = Field( + default=None, + description='Optional. Metadata associated with the artifact update.', + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class AgentInterface(A2ABaseModel): """Declares a combination of a target URL, transport and protocol version for interacting with the agent. This allows agents to expose the same functionality over multiple protocol binding mechanisms.""" - url: str = Field(..., description='The URL where this interface is available. Must be a valid absolute HTTPS URL in production. Example: "https://api.example.com/a2a/v1", "https://grpc.example.com/a2a"') - protocol_binding: str = Field(..., description='The protocol binding supported at this URL. This is an open form string, to be easily extended for other protocol bindings. The core ones officially supported are `JSONRPC`, `GRPC` and `HTTP+JSON`.') - tenant: str = Field(default='', description='Tenant ID to be used in the request when calling the agent.') - protocol_version: str = Field(..., description='The version of the A2A protocol this interface exposes. Use the latest supported minor version per major version. Examples: "0.3", "1.0"') + url: str = Field( + ..., + description='The URL where this interface is available. Must be a valid absolute HTTPS URL in production. Example: "https://api.example.com/a2a/v1", "https://grpc.example.com/a2a"', + ) + protocol_binding: str = Field( + ..., + description='The protocol binding supported at this URL. This is an open form string, to be easily extended for other protocol bindings. The core ones officially supported are `JSONRPC`, `GRPC` and `HTTP+JSON`.', + ) + tenant: str = Field( + default='', + description='Tenant ID to be used in the request when calling the agent.', + ) + protocol_version: str = Field( + ..., + description='The version of the A2A protocol this interface exposes. Use the latest supported minor version per major version. Examples: "0.3", "1.0"', + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class AgentProvider(A2ABaseModel): """Represents the service provider of an agent.""" - url: str = Field(..., description="A URL for the agent provider's website or relevant documentation. Example: \"https://ai.google.dev\"") - organization: str = Field(..., description="The name of the agent provider's organization. Example: \"Google\"") + url: str = Field( + ..., + description='A URL for the agent provider\'s website or relevant documentation. Example: "https://ai.google.dev"', + ) + organization: str = Field( + ..., + description='The name of the agent provider\'s organization. Example: "Google"', + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class AgentCapabilities(A2ABaseModel): """Defines optional capabilities supported by an agent.""" - streaming: bool | None = Field(default=None, description='Indicates if the agent supports streaming responses.') - push_notifications: bool | None = Field(default=None, description='Indicates if the agent supports sending push notifications for asynchronous task updates.') - extensions: list[AgentExtension] | None = Field(default=None, description='A list of protocol extensions supported by the agent.') - extended_agent_card: bool | None = Field(default=None, description='Indicates if the agent supports providing an extended agent card when authenticated.') + streaming: bool | None = Field( + default=None, + description='Indicates if the agent supports streaming responses.', + ) + push_notifications: bool | None = Field( + default=None, + description='Indicates if the agent supports sending push notifications for asynchronous task updates.', + ) + extensions: list[AgentExtension] | None = Field( + default=None, + description='A list of protocol extensions supported by the agent.', + ) + extended_agent_card: bool | None = Field( + default=None, + description='Indicates if the agent supports providing an extended agent card when authenticated.', + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class AgentSkill(A2ABaseModel): """Represents a distinct capability or function that an agent can perform.""" - id: str = Field(..., description="A unique identifier for the agent's skill.") + id: str = Field( + ..., description="A unique identifier for the agent's skill." + ) name: str = Field(..., description='A human-readable name for the skill.') - description: str = Field(..., description='A detailed description of the skill.') - tags: list[str] = Field(..., description="A set of keywords describing the skill's capabilities.") - examples: list[str] | None = Field(default=None, description='Example prompts or scenarios that this skill can handle.') - input_modes: list[str] | None = Field(default=None, description="The set of supported input media types for this skill, overriding the agent's defaults.") - output_modes: list[str] | None = Field(default=None, description="The set of supported output media types for this skill, overriding the agent's defaults.") - security_requirements: list[SecurityRequirement] | None = Field(default=None, description='Security schemes necessary for this skill.') + description: str = Field( + ..., description='A detailed description of the skill.' + ) + tags: list[str] = Field( + ..., + description="A set of keywords describing the skill's capabilities.", + ) + examples: list[str] | None = Field( + default=None, + description='Example prompts or scenarios that this skill can handle.', + ) + input_modes: list[str] | None = Field( + default=None, + description="The set of supported input media types for this skill, overriding the agent's defaults.", + ) + output_modes: list[str] | None = Field( + default=None, + description="The set of supported output media types for this skill, overriding the agent's defaults.", + ) + security_requirements: list[SecurityRequirement] | None = Field( + default=None, description='Security schemes necessary for this skill.' + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class AgentCard(A2ABaseModel): """A self-describing manifest for an agent. It provides essential metadata including the agent's identity, capabilities, skills, supported communication methods, and security requirements. Next ID: 20.""" - name: str = Field(..., description='A human readable name for the agent. Example: "Recipe Agent"') - description: str = Field(..., description='A human-readable description of the agent, assisting users and other agents in understanding its purpose. Example: "Agent that helps users with recipes and cooking."') - supported_interfaces: list[AgentInterface] = Field(..., description='Ordered list of supported interfaces. The first entry is preferred.') - provider: AgentProvider | None = Field(default=None, description='The service provider of the agent.') - version: str = Field(..., description='The version of the agent. Example: "1.0.0"') - documentation_url: str | None = Field(default=None, description='A URL providing additional documentation about the agent.') - capabilities: AgentCapabilities = Field(..., description='A2A Capability set supported by the agent.') - security_schemes: dict[str, SecurityScheme] | None = Field(default=None, description='The security scheme details used for authenticating with this agent.') - security_requirements: list[SecurityRequirement] | None = Field(default=None, description='Security requirements for contacting the agent.') - default_input_modes: list[str] = Field(..., description='protolint:enable REPEATED_FIELD_NAMES_PLURALIZED The set of interaction modes that the agent supports across all skills. This can be overridden per skill. Defined as media types.') - default_output_modes: list[str] = Field(..., description='The media types supported as outputs from this agent.') - skills: list[AgentSkill] = Field(..., description='Skills represent the abilities of an agent. It is largely a descriptive concept but represents a more focused set of behaviors that the agent is likely to succeed at.') - signatures: list[AgentCardSignature] | None = Field(default=None, description='JSON Web Signatures computed for this `AgentCard`.') - icon_url: str | None = Field(default=None, description='Optional. A URL to an icon for the agent.') + name: str = Field( + ..., + description='A human readable name for the agent. Example: "Recipe Agent"', + ) + description: str = Field( + ..., + description='A human-readable description of the agent, assisting users and other agents in understanding its purpose. Example: "Agent that helps users with recipes and cooking."', + ) + supported_interfaces: list[AgentInterface] = Field( + ..., + description='Ordered list of supported interfaces. The first entry is preferred.', + ) + provider: AgentProvider | None = Field( + default=None, description='The service provider of the agent.' + ) + version: str = Field( + ..., description='The version of the agent. Example: "1.0.0"' + ) + documentation_url: str | None = Field( + default=None, + description='A URL providing additional documentation about the agent.', + ) + capabilities: AgentCapabilities = Field( + ..., description='A2A Capability set supported by the agent.' + ) + security_schemes: dict[str, SecurityScheme] | None = Field( + default=None, + description='The security scheme details used for authenticating with this agent.', + ) + security_requirements: list[SecurityRequirement] | None = Field( + default=None, + description='Security requirements for contacting the agent.', + ) + default_input_modes: list[str] = Field( + ..., + description='protolint:enable REPEATED_FIELD_NAMES_PLURALIZED The set of interaction modes that the agent supports across all skills. This can be overridden per skill. Defined as media types.', + ) + default_output_modes: list[str] = Field( + ..., description='The media types supported as outputs from this agent.' + ) + skills: list[AgentSkill] = Field( + ..., + description='Skills represent the abilities of an agent. It is largely a descriptive concept but represents a more focused set of behaviors that the agent is likely to succeed at.', + ) + signatures: list[AgentCardSignature] | None = Field( + default=None, + description='JSON Web Signatures computed for this `AgentCard`.', + ) + icon_url: str | None = Field( + default=None, description='Optional. A URL to an icon for the agent.' + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class AgentExtension(A2ABaseModel): """A declaration of a protocol extension supported by an Agent.""" - uri: str = Field(default='', description='The unique URI identifying the extension.') - description: str = Field(default='', description='A human-readable description of how this agent uses the extension.') - required: bool = Field(default=False, description="If true, the client must understand and comply with the extension's requirements.") - params: dict[str, Any] | None = Field(default=None, description='Optional. Extension-specific configuration parameters.') + uri: str = Field( + default='', description='The unique URI identifying the extension.' + ) + description: str = Field( + default='', + description='A human-readable description of how this agent uses the extension.', + ) + required: bool = Field( + default=False, + description="If true, the client must understand and comply with the extension's requirements.", + ) + params: dict[str, Any] | None = Field( + default=None, + description='Optional. Extension-specific configuration parameters.', + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class AgentCardSignature(A2ABaseModel): """AgentCardSignature represents a JWS signature of an AgentCard. This follows the JSON format of an RFC 7515 JSON Web Signature (JWS).""" - protected: str = Field(..., description='(-- api-linter: core::0140::reserved-words=disabled aip.dev/not-precedent: Backwards compatibility --) Required. The protected JWS header for the signature. This is always a base64url-encoded JSON object.') - signature: str = Field(..., description='Required. The computed signature, base64url-encoded.') - header: dict[str, Any] | None = Field(default=None, description='The unprotected JWS header values.') + protected: str = Field( + ..., + description='(-- api-linter: core::0140::reserved-words=disabled aip.dev/not-precedent: Backwards compatibility --) Required. The protected JWS header for the signature. This is always a base64url-encoded JSON object.', + ) + signature: str = Field( + ..., description='Required. The computed signature, base64url-encoded.' + ) + header: dict[str, Any] | None = Field( + default=None, description='The unprotected JWS header values.' + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class StringList(A2ABaseModel): """protolint:disable REPEATED_FIELD_NAMES_PLURALIZED A list of strings.""" - list_: list[str] | None = Field(default=None, alias='list', description='The individual string values.') + list_: list[str] | None = Field( + default=None, alias='list', description='The individual string values.' + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class SecurityRequirement(A2ABaseModel): """Defines the security requirements for an agent.""" - schemes: dict[str, StringList] | None = Field(default=None, description='A map of security schemes to the required scopes.') + schemes: dict[str, StringList] | None = Field( + default=None, + description='A map of security schemes to the required scopes.', + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class SecurityScheme(A2ABaseModel): """Defines a security scheme that can be used to secure an agent's endpoints. This is a discriminated union type based on the OpenAPI 3.2 Security Scheme Object. See: https://spec.openapis.org/oas/v3.2.0.html#security-scheme-object.""" - scheme: APIKeySecurityScheme | HTTPAuthSecurityScheme | OAuth2SecurityScheme | OpenIdConnectSecurityScheme | MutualTlsSecurityScheme | None = None - + scheme: ( + APIKeySecurityScheme + | HTTPAuthSecurityScheme + | OAuth2SecurityScheme + | OpenIdConnectSecurityScheme + | MutualTlsSecurityScheme + | None + ) = None def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class APIKeySecurityScheme(A2ABaseModel): """Defines a security scheme using an API key.""" - description: str = Field(default='', description='An optional description for the security scheme.') - location: str = Field(..., description='The location of the API key. Valid values are "query", "header", or "cookie".') - name: str = Field(..., description='The name of the header, query, or cookie parameter to be used.') + description: str = Field( + default='', + description='An optional description for the security scheme.', + ) + location: str = Field( + ..., + description='The location of the API key. Valid values are "query", "header", or "cookie".', + ) + name: str = Field( + ..., + description='The name of the header, query, or cookie parameter to be used.', + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class HTTPAuthSecurityScheme(A2ABaseModel): """Defines a security scheme using HTTP authentication.""" - description: str = Field(default='', description='An optional description for the security scheme.') - scheme: str = Field(..., description='The name of the HTTP Authentication scheme to be used in the Authorization header, as defined in RFC7235 (e.g., "Bearer"). This value should be registered in the IANA Authentication Scheme registry.') - bearer_format: str = Field(default='', description='A hint to the client to identify how the bearer token is formatted (e.g., "JWT"). Primarily for documentation purposes.') + description: str = Field( + default='', + description='An optional description for the security scheme.', + ) + scheme: str = Field( + ..., + description='The name of the HTTP Authentication scheme to be used in the Authorization header, as defined in RFC7235 (e.g., "Bearer"). This value should be registered in the IANA Authentication Scheme registry.', + ) + bearer_format: str = Field( + default='', + description='A hint to the client to identify how the bearer token is formatted (e.g., "JWT"). Primarily for documentation purposes.', + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class OAuthFlows(A2ABaseModel): """Defines the configuration for the supported OAuth 2.0 flows.""" - flow: AuthorizationCodeOAuthFlow | ClientCredentialsOAuthFlow | ImplicitOAuthFlow | PasswordOAuthFlow | DeviceCodeOAuthFlow | None = None - + flow: ( + AuthorizationCodeOAuthFlow + | ClientCredentialsOAuthFlow + | ImplicitOAuthFlow + | PasswordOAuthFlow + | DeviceCodeOAuthFlow + | None + ) = None def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class OAuth2SecurityScheme(A2ABaseModel): """Defines a security scheme using OAuth 2.0.""" - description: str = Field(default='', description='An optional description for the security scheme.') - flows: OAuthFlows = Field(..., description='An object containing configuration information for the supported OAuth 2.0 flows.') - oauth2_metadata_url: str = Field(default='', description='URL to the OAuth2 authorization server metadata [RFC 8414](https://datatracker.ietf.org/doc/html/rfc8414). TLS is required.') + description: str = Field( + default='', + description='An optional description for the security scheme.', + ) + flows: OAuthFlows = Field( + ..., + description='An object containing configuration information for the supported OAuth 2.0 flows.', + ) + oauth2_metadata_url: str = Field( + default='', + description='URL to the OAuth2 authorization server metadata [RFC 8414](https://datatracker.ietf.org/doc/html/rfc8414). TLS is required.', + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class OpenIdConnectSecurityScheme(A2ABaseModel): """Defines a security scheme using OpenID Connect.""" - description: str = Field(default='', description='An optional description for the security scheme.') - open_id_connect_url: str = Field(..., description="The [OpenID Connect Discovery URL](https://openid.net/specs/openid-connect-discovery-1_0.html) for the OIDC provider's metadata.") + description: str = Field( + default='', + description='An optional description for the security scheme.', + ) + open_id_connect_url: str = Field( + ..., + description="The [OpenID Connect Discovery URL](https://openid.net/specs/openid-connect-discovery-1_0.html) for the OIDC provider's metadata.", + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class MutualTlsSecurityScheme(A2ABaseModel): """Defines a security scheme using mTLS authentication.""" - description: str = Field(default='', description='An optional description for the security scheme.') + description: str = Field( + default='', + description='An optional description for the security scheme.', + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class AuthorizationCodeOAuthFlow(A2ABaseModel): """Defines configuration details for the OAuth 2.0 Authorization Code flow.""" - authorization_url: str = Field(..., description='The authorization URL to be used for this flow.') - token_url: str = Field(..., description='The token URL to be used for this flow.') - refresh_url: str = Field(default='', description='The URL to be used for obtaining refresh tokens.') - scopes: dict[str, str] = Field(..., description='The available scopes for the OAuth2 security scheme.') - pkce_required: bool = Field(default=False, description='Indicates if PKCE (RFC 7636) is required for this flow. PKCE should always be used for public clients and is recommended for all clients.') + authorization_url: str = Field( + ..., description='The authorization URL to be used for this flow.' + ) + token_url: str = Field( + ..., description='The token URL to be used for this flow.' + ) + refresh_url: str = Field( + default='', + description='The URL to be used for obtaining refresh tokens.', + ) + scopes: dict[str, str] = Field( + ..., description='The available scopes for the OAuth2 security scheme.' + ) + pkce_required: bool = Field( + default=False, + description='Indicates if PKCE (RFC 7636) is required for this flow. PKCE should always be used for public clients and is recommended for all clients.', + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class ClientCredentialsOAuthFlow(A2ABaseModel): """Defines configuration details for the OAuth 2.0 Client Credentials flow.""" - token_url: str = Field(..., description='The token URL to be used for this flow.') - refresh_url: str = Field(default='', description='The URL to be used for obtaining refresh tokens.') - scopes: dict[str, str] = Field(..., description='The available scopes for the OAuth2 security scheme.') + token_url: str = Field( + ..., description='The token URL to be used for this flow.' + ) + refresh_url: str = Field( + default='', + description='The URL to be used for obtaining refresh tokens.', + ) + scopes: dict[str, str] = Field( + ..., description='The available scopes for the OAuth2 security scheme.' + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class ImplicitOAuthFlow(A2ABaseModel): """Deprecated: Use Authorization Code + PKCE instead.""" - authorization_url: str = Field(default='', description='The authorization URL to be used for this flow. This MUST be in the form of a URL. The OAuth2 standard requires the use of TLS') - refresh_url: str = Field(default='', description='The URL to be used for obtaining refresh tokens. This MUST be in the form of a URL. The OAuth2 standard requires the use of TLS.') - scopes: dict[str, str] | None = Field(default=None, description='The available scopes for the OAuth2 security scheme. A map between the scope name and a short description for it. The map MAY be empty.') + authorization_url: str = Field( + default='', + description='The authorization URL to be used for this flow. This MUST be in the form of a URL. The OAuth2 standard requires the use of TLS', + ) + refresh_url: str = Field( + default='', + description='The URL to be used for obtaining refresh tokens. This MUST be in the form of a URL. The OAuth2 standard requires the use of TLS.', + ) + scopes: dict[str, str] | None = Field( + default=None, + description='The available scopes for the OAuth2 security scheme. A map between the scope name and a short description for it. The map MAY be empty.', + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class PasswordOAuthFlow(A2ABaseModel): """Deprecated: Use Authorization Code + PKCE or Device Code.""" - token_url: str = Field(default='', description='The token URL to be used for this flow. This MUST be in the form of a URL. The OAuth2 standard requires the use of TLS.') - refresh_url: str = Field(default='', description='The URL to be used for obtaining refresh tokens. This MUST be in the form of a URL. The OAuth2 standard requires the use of TLS.') - scopes: dict[str, str] | None = Field(default=None, description='The available scopes for the OAuth2 security scheme. A map between the scope name and a short description for it. The map MAY be empty.') + token_url: str = Field( + default='', + description='The token URL to be used for this flow. This MUST be in the form of a URL. The OAuth2 standard requires the use of TLS.', + ) + refresh_url: str = Field( + default='', + description='The URL to be used for obtaining refresh tokens. This MUST be in the form of a URL. The OAuth2 standard requires the use of TLS.', + ) + scopes: dict[str, str] | None = Field( + default=None, + description='The available scopes for the OAuth2 security scheme. A map between the scope name and a short description for it. The map MAY be empty.', + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class DeviceCodeOAuthFlow(A2ABaseModel): """Defines configuration details for the OAuth 2.0 Device Code flow (RFC 8628). This flow is designed for input-constrained devices such as IoT devices, and CLI tools where the user authenticates on a separate device.""" - device_authorization_url: str = Field(..., description='The device authorization endpoint URL.') - token_url: str = Field(..., description='The token URL to be used for this flow.') - refresh_url: str = Field(default='', description='The URL to be used for obtaining refresh tokens.') - scopes: dict[str, str] = Field(..., description='The available scopes for the OAuth2 security scheme.') + device_authorization_url: str = Field( + ..., description='The device authorization endpoint URL.' + ) + token_url: str = Field( + ..., description='The token URL to be used for this flow.' + ) + refresh_url: str = Field( + default='', + description='The URL to be used for obtaining refresh tokens.', + ) + scopes: dict[str, str] = Field( + ..., description='The available scopes for the OAuth2 security scheme.' + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class SendMessageRequest(A2ABaseModel): """Represents a request for the `SendMessage` method.""" - tenant: str = Field(default='', description='Optional. Tenant ID, provided as a path parameter.') - message: Message = Field(..., description='The message to send to the agent.') - configuration: SendMessageConfiguration | None = Field(default=None, description='Configuration for the send request.') - metadata: dict[str, Any] | None = Field(default=None, description='A flexible key-value map for passing additional context or parameters.') + tenant: str = Field( + default='', + description='Optional. Tenant ID, provided as a path parameter.', + ) + message: Message = Field( + ..., description='The message to send to the agent.' + ) + configuration: SendMessageConfiguration | None = Field( + default=None, description='Configuration for the send request.' + ) + metadata: dict[str, Any] | None = Field( + default=None, + description='A flexible key-value map for passing additional context or parameters.', + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class GetTaskRequest(A2ABaseModel): """Represents a request for the `GetTask` method.""" - tenant: str = Field(default='', description='Optional. Tenant ID, provided as a path parameter.') + tenant: str = Field( + default='', + description='Optional. Tenant ID, provided as a path parameter.', + ) id: str = Field(..., description='The resource ID of the task to retrieve.') - history_length: int | None = Field(default=None, description="The maximum number of most recent messages from the task's history to retrieve. An unset value means the client does not impose any limit. A value of zero is a request to not include any messages. The server MUST NOT return more messages than the provided value, but MAY apply a lower limit.") + history_length: int | None = Field( + default=None, + description="The maximum number of most recent messages from the task's history to retrieve. An unset value means the client does not impose any limit. A value of zero is a request to not include any messages. The server MUST NOT return more messages than the provided value, but MAY apply a lower limit.", + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class ListTasksRequest(A2ABaseModel): """Parameters for listing tasks with optional filtering criteria.""" - tenant: str = Field(default='', description='Tenant ID, provided as a path parameter.') - context_id: str = Field(default='', description='Filter tasks by context ID to get tasks from a specific conversation or session.') - status: TaskState = Field(default=TaskState.TASK_STATE_UNSPECIFIED, description='Filter tasks by their current status state.') - page_size: int | None = Field(default=None, description='The maximum number of tasks to return. The service may return fewer than this value. If unspecified, at most 50 tasks will be returned. The minimum value is 1. The maximum value is 100.') - page_token: str = Field(default='', description='A page token, received from a previous `ListTasks` call. `ListTasksResponse.next_page_token`. Provide this to retrieve the subsequent page.') - history_length: int | None = Field(default=None, description="The maximum number of messages to include in each task's history.") - status_timestamp_after: datetime | None = Field(default=None, description='Filter tasks which have a status updated after the provided timestamp in ISO 8601 format (e.g., "2023-10-27T10:00:00Z"). Only tasks with a status timestamp time greater than or equal to this value will be returned.') - include_artifacts: bool | None = Field(default=None, description='Whether to include artifacts in the returned tasks. Defaults to false to reduce payload size.') + tenant: str = Field( + default='', description='Tenant ID, provided as a path parameter.' + ) + context_id: str = Field( + default='', + description='Filter tasks by context ID to get tasks from a specific conversation or session.', + ) + status: TaskState = Field( + default=TaskState.TASK_STATE_UNSPECIFIED, + description='Filter tasks by their current status state.', + ) + page_size: int | None = Field( + default=None, + description='The maximum number of tasks to return. The service may return fewer than this value. If unspecified, at most 50 tasks will be returned. The minimum value is 1. The maximum value is 100.', + ) + page_token: str = Field( + default='', + description='A page token, received from a previous `ListTasks` call. `ListTasksResponse.next_page_token`. Provide this to retrieve the subsequent page.', + ) + history_length: int | None = Field( + default=None, + description="The maximum number of messages to include in each task's history.", + ) + status_timestamp_after: datetime | None = Field( + default=None, + description='Filter tasks which have a status updated after the provided timestamp in ISO 8601 format (e.g., "2023-10-27T10:00:00Z"). Only tasks with a status timestamp time greater than or equal to this value will be returned.', + ) + include_artifacts: bool | None = Field( + default=None, + description='Whether to include artifacts in the returned tasks. Defaults to false to reduce payload size.', + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" @@ -528,137 +860,176 @@ def _serialize_timestamp(cls, v: datetime | None, _info: Any) -> str | None: """Serialize datetime to RFC 3339 with UTC 'Z' suffix for ProtoJSON.""" if v is None: return None - return v.strftime('%Y-%m-%dT%H:%M:%S.') + f'{v.microsecond // 1000:03d}' + 'Z' - + return ( + v.strftime('%Y-%m-%dT%H:%M:%S.') + + f'{v.microsecond // 1000:03d}' + + 'Z' + ) class ListTasksResponse(A2ABaseModel): """Result object for `ListTasks` method containing an array of tasks and pagination information.""" - tasks: list[Task] = Field(..., description='Array of tasks matching the specified criteria.') - next_page_token: str = Field(..., description='A token to retrieve the next page of results, or empty if there are no more results in the list.') - page_size: int = Field(..., description='The page size used for this response.') - total_size: int = Field(..., description='Total number of tasks available (before pagination).') + tasks: list[Task] = Field( + ..., description='Array of tasks matching the specified criteria.' + ) + next_page_token: str = Field( + ..., + description='A token to retrieve the next page of results, or empty if there are no more results in the list.', + ) + page_size: int = Field( + ..., description='The page size used for this response.' + ) + total_size: int = Field( + ..., description='Total number of tasks available (before pagination).' + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class CancelTaskRequest(A2ABaseModel): """Represents a request for the `CancelTask` method.""" - tenant: str = Field(default='', description='Optional. Tenant ID, provided as a path parameter.') + tenant: str = Field( + default='', + description='Optional. Tenant ID, provided as a path parameter.', + ) id: str = Field(..., description='The resource ID of the task to cancel.') - metadata: dict[str, Any] | None = Field(default=None, description='A flexible key-value map for passing additional context or parameters.') + metadata: dict[str, Any] | None = Field( + default=None, + description='A flexible key-value map for passing additional context or parameters.', + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class GetTaskPushNotificationConfigRequest(A2ABaseModel): """Represents a request for the `GetTaskPushNotificationConfig` method.""" - tenant: str = Field(default='', description='Optional. Tenant ID, provided as a path parameter.') + tenant: str = Field( + default='', + description='Optional. Tenant ID, provided as a path parameter.', + ) task_id: str = Field(..., description='The parent task resource ID.') - id: str = Field(..., description='The resource ID of the configuration to retrieve.') + id: str = Field( + ..., description='The resource ID of the configuration to retrieve.' + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class DeleteTaskPushNotificationConfigRequest(A2ABaseModel): """Represents a request for the `DeleteTaskPushNotificationConfig` method.""" - tenant: str = Field(default='', description='Optional. Tenant ID, provided as a path parameter.') + tenant: str = Field( + default='', + description='Optional. Tenant ID, provided as a path parameter.', + ) task_id: str = Field(..., description='The parent task resource ID.') - id: str = Field(..., description='The resource ID of the configuration to delete.') + id: str = Field( + ..., description='The resource ID of the configuration to delete.' + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class SubscribeToTaskRequest(A2ABaseModel): """Represents a request for the `SubscribeToTask` method.""" - tenant: str = Field(default='', description='Optional. Tenant ID, provided as a path parameter.') - id: str = Field(..., description='The resource ID of the task to subscribe to.') + tenant: str = Field( + default='', + description='Optional. Tenant ID, provided as a path parameter.', + ) + id: str = Field( + ..., description='The resource ID of the task to subscribe to.' + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class ListTaskPushNotificationConfigsRequest(A2ABaseModel): """Represents a request for the `ListTaskPushNotificationConfigs` method.""" - tenant: str = Field(default='', description='Optional. Tenant ID, provided as a path parameter.') + tenant: str = Field( + default='', + description='Optional. Tenant ID, provided as a path parameter.', + ) task_id: str = Field(..., description='The parent task resource ID.') - page_size: int = Field(default=0, description='The maximum number of configurations to return.') - page_token: str = Field(default='', description='A page token received from a previous `ListTaskPushNotificationConfigsRequest` call.') + page_size: int = Field( + default=0, description='The maximum number of configurations to return.' + ) + page_token: str = Field( + default='', + description='A page token received from a previous `ListTaskPushNotificationConfigsRequest` call.', + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class GetExtendedAgentCardRequest(A2ABaseModel): """Represents a request for the `GetExtendedAgentCard` method.""" - tenant: str = Field(default='', description='Optional. Tenant ID, provided as a path parameter.') + tenant: str = Field( + default='', + description='Optional. Tenant ID, provided as a path parameter.', + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class SendMessageResponse(A2ABaseModel): """Represents the response for the `SendMessage` method.""" payload: Task | Message | None = None - def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class StreamResponse(A2ABaseModel): """A wrapper object used in streaming operations to encapsulate different types of response data.""" - payload: Task | Message | TaskStatusUpdateEvent | TaskArtifactUpdateEvent | None = None - + payload: ( + Task | Message | TaskStatusUpdateEvent | TaskArtifactUpdateEvent | None + ) = None def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - class ListTaskPushNotificationConfigsResponse(A2ABaseModel): """Represents a successful response for the `ListTaskPushNotificationConfigs` method.""" - configs: list[TaskPushNotificationConfig] | None = Field(default=None, description='The list of push notification configurations.') - next_page_token: str = Field(default='', description='A token to retrieve the next page of results, or empty if there are no more results in the list.') + configs: list[TaskPushNotificationConfig] | None = Field( + default=None, + description='The list of push notification configurations.', + ) + next_page_token: str = Field( + default='', + description='A token to retrieve the next page of results, or empty if there are no more results in the list.', + ) def to_proto_json(self) -> dict: """Serialize to a ProtoJSON-compatible dict (camelCase keys, no None values).""" return self.model_dump(by_alias=True, exclude_none=True) - - __all__ = [ 'APIKeySecurityScheme', 'AgentCapabilities',