diff --git a/.github/actions/spelling/allow.txt b/.github/actions/spelling/allow.txt index df74a242..5e5c61f4 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 ed59a649..3d15e2ed 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"] diff --git a/buf.gen.yaml b/buf.gen.yaml index d7937469..37333a2d 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 34ff96ae..036fb5e3 100755 --- a/scripts/gen_proto.sh +++ b/scripts/gen_proto.sh @@ -1,9 +1,17 @@ #!/bin/bash set -e +# Install proto2pydantic plugin for Pydantic model generation +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 +# 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 new file mode 100644 index 00000000..b2466324 --- /dev/null +++ b/src/a2a/types/a2a_pydantic.py @@ -0,0 +1,1085 @@ +"""Generated by proto2pydantic from a2a.proto. DO NOT EDIT.""" + +from __future__ import annotations + +from enum import Enum +from typing import TYPE_CHECKING, Any + + +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`.""" + + 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 | 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.', + ) + + 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.', + ) + 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.', + ) + 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.', + ) + + 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"', + ) + + 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 // 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.', + ) + + 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.', + ) + + 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] | 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.', + ) + + 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 | 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.', + ) + + 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.' + ) + + 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 | 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.', + ) + 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=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).""" + 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 // 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).' + ) + + 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] | 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.', + ) + 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__ = [ + '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()