fasta2a
Bases: ABC, Generic[ContextT]
A storage to retrieve and save tasks, as well as retrieve and save context.
The storage serves two purposes:
- Task storage: Stores tasks in A2A protocol format with their status, artifacts, and message history
- Context storage: Stores conversation context in a format optimized for the specific agent implementation
@abstractmethod
@async
def load_task(task_id: str, history_length: int | None = None) -> Task | None
Load a task from storage.
If the task is not found, return None.
Task | None
@abstractmethod
@async
def submit_task(context_id: str, message: Message) -> Task
Submit a task to storage.
Task
@abstractmethod
@async
def update_task(
task_id: str,
state: TaskState,
new_artifacts: list[Artifact] | None = None,
new_messages: list[Message] | None = None,
) -> Task
Update the state of a task. Appends artifacts and messages, if specified.
Task
@abstractmethod
@async
def load_context(context_id: str) -> ContextT | None
Retrieve the stored context given the context_id.
ContextT | None
@abstractmethod
@async
def update_context(context_id: str, context: ContextT) -> None
Updates the context for a context_id.
Implementing agent can decide what to store in context.
Bases: ABC
The broker class is in charge of scheduling the tasks.
The HTTP server uses the broker to schedule tasks.
The simple implementation is the InMemoryBroker, which is the broker that
runs the tasks in the same process as the HTTP server. That said, this class can be
extended to support remote workers.
@abstractmethod
@async
def run_task(params: TaskSendParams) -> None
Send a task to be executed by the worker.
@abstractmethod
@async
def cancel_task(params: TaskIdParams) -> None
Cancel a task.
@abstractmethod
def receive_task_operations() -> AsyncIterator[TaskOperation]
Receive task operations from the broker.
On a multi-worker setup, the broker will need to round-robin the task operations between the workers.
AsyncIterator[TaskOperation]
Bases: ABC, Generic[ContextT]
A worker is responsible for executing tasks.
@async
def run() -> AsyncIterator[None]
Run the worker.
It connects to the broker, and it makes itself available to receive commands.
Bases: Starlette
The main class for the FastA2A library.
Bases: TypedDict
Skills are a unit of capability that an agent can perform.
A unique identifier for the skill.
Type: str
Human readable name of the skill.
Type: str
A human-readable description of the skill.
It will be used by the client or a human as a hint to understand the skill.
Type: str
Set of tag-words describing classes of capabilities for this specific skill.
Examples: “cooking”, “customer support”, “billing”.
The set of example scenarios that the skill can perform.
Will be used by the client as a hint to understand how the skill can be used. (e.g. “I need a recipe for bread”)
Type: NotRequired[list[str]]
Supported mime types for input data.
Supported mime types for output data.
This module contains the schema for the agent card.
Bases: TypedDict
The card that describes an agent.
Human readable name of the agent e.g. “Recipe Agent”.
Type: str
A human-readable description of the agent.
Used to assist users and other agents in understanding what the agent can do. (e.g. “Agent that helps users with recipes and cooking.”)
Type: str
A URL to the address the agent is hosted at.
Type: str
The version of the agent - format is up to the provider. (e.g. “1.0.0”)
Type: str
The version of the A2A protocol this agent supports.
Type: str
The service provider of the agent.
Type: NotRequired[AgentProvider]
A URL to documentation for the agent.
Type: NotRequired[str]
A URL to an icon for the agent.
Type: NotRequired[str]
The transport of the preferred endpoint. If empty, defaults to JSONRPC.
Type: NotRequired[str]
Announcement of additional supported transports.
Type: NotRequired[list[AgentInterface]]
The capabilities of the agent.
Type: AgentCapabilities
Security requirements for contacting the agent.
Type: NotRequired[list[dict[str, list[str]]]]
Security scheme definitions.
Type: NotRequired[dict[str, SecurityScheme]]
Supported mime types for input data.
Supported mime types for output data.
The set of skills, or distinct capabilities, that the agent can perform.
Type: list[Skill]
Bases: TypedDict
The service provider of the agent.
The name of the agent provider’s organization.
Type: str
A URL for the agent provider’s website or relevant documentation.
Type: str
Bases: TypedDict
The capabilities of the agent.
Whether the agent supports streaming.
Type: NotRequired[bool]
Whether the agent can notify updates to client.
Type: NotRequired[bool]
Whether the agent exposes status change history for tasks.
Type: NotRequired[bool]
Bases: TypedDict
HTTP security scheme.
The type of the security scheme. Must be ‘http’.
Type: Literal[‘http’]
The name of the HTTP Authorization scheme.
Type: str
A hint to the client to identify how the bearer token is formatted.
Type: NotRequired[str]
Description of this security scheme.
Type: NotRequired[str]
Bases: TypedDict
API Key security scheme.
The type of the security scheme. Must be ‘apiKey’.
Type: Literal[‘apiKey’]
The name of the header, query or cookie parameter to be used.
Type: str
The location of the API key.
Type: Literal[‘query’, ‘header’, ‘cookie’]
Description of this security scheme.
Type: NotRequired[str]
Bases: TypedDict
OAuth2 security scheme.
The type of the security scheme. Must be ‘oauth2’.
Type: Literal[‘oauth2’]
An object containing configuration information for the flow types supported.
Description of this security scheme.
Type: NotRequired[str]
Bases: TypedDict
OpenID Connect security scheme.
The type of the security scheme. Must be ‘openIdConnect’.
Type: Literal[‘openIdConnect’]
OpenId Connect URL to discover OAuth2 configuration values.
Type: str
Description of this security scheme.
Type: NotRequired[str]
Bases: TypedDict
An interface that the agent supports.
The transport protocol (e.g., ‘jsonrpc’, ‘websocket’).
Type: str
The URL endpoint for this transport.
Type: str
Description of this interface.
Type: NotRequired[str]
Bases: TypedDict
A declaration of an extension supported by an Agent.
The URI of the extension.
Type: str
A description of how this agent uses this extension.
Type: NotRequired[str]
Whether the client must follow specific requirements of the extension.
Type: NotRequired[bool]
Optional configuration for the extension.
Type: NotRequired[dict[str, Any]]
Bases: TypedDict
Skills are a unit of capability that an agent can perform.
A unique identifier for the skill.
Type: str
Human readable name of the skill.
Type: str
A human-readable description of the skill.
It will be used by the client or a human as a hint to understand the skill.
Type: str
Set of tag-words describing classes of capabilities for this specific skill.
Examples: “cooking”, “customer support”, “billing”.
The set of example scenarios that the skill can perform.
Will be used by the client as a hint to understand how the skill can be used. (e.g. “I need a recipe for bread”)
Type: NotRequired[list[str]]
Supported mime types for input data.
Supported mime types for output data.
Bases: TypedDict
Agents generate Artifacts as an end result of a Task.
Artifacts are immutable, can be named, and can have multiple parts. A streaming response can append parts to existing Artifacts.
A single Task can generate many Artifacts. For example, “create a webpage” could create separate HTML and image Artifacts.
Unique identifier for the artifact.
Type: str
The name of the artifact.
Type: NotRequired[str]
A description of the artifact.
Type: NotRequired[str]
The parts that make up the artifact.
Type: list[Part]
Metadata about the artifact.
Type: NotRequired[dict[str, Any]]
Array of extensions.
Type: NotRequired[list[str]]
Whether to append this artifact to an existing one.
Type: NotRequired[bool]
Whether this is the last chunk of the artifact.
Type: NotRequired[bool]
Bases: TypedDict
Configuration for push notifications.
A2A supports a secure notification mechanism whereby an agent can notify a client of an update outside a connected session via a PushNotificationService. Within and across enterprises, it is critical that the agent verifies the identity of the notification service, authenticates itself with the service, and presents an identifier that ties the notification to the executing Task.
The target server of the PushNotificationService should be considered a separate service, and is not guaranteed (or even expected) to be the client directly. This PushNotificationService is responsible for authenticating and authorizing the agent and for proxying the verified notification to the appropriate endpoint (which could be anything from a pub/sub queue, to an email inbox or other service, etc.).
For contrived scenarios with isolated client-agent pairs (e.g. local service mesh in a contained VPC, etc.) or isolated environments without enterprise security concerns, the client may choose to simply open a port and act as its own PushNotificationService. Any enterprise implementation will likely have a centralized service that authenticates the remote agents with trusted notification credentials and can handle online/offline scenarios. (This should be thought of similarly to a mobile Push Notification Service).
Server-assigned identifier.
Type: NotRequired[str]
The URL to send push notifications to.
Type: str
Token unique to this task/session.
Type: NotRequired[str]
Authentication details for push notifications.
Type: NotRequired[SecurityScheme]
Bases: TypedDict
Configuration for task push notifications.
The task id.
Type: str
The push notification configuration.
Type: PushNotificationConfig
Bases: TypedDict
A Message contains any content that is not an Artifact.
This can include things like agent thoughts, user context, instructions, errors, status, or metadata.
All content from a client comes in the form of a Message. Agents send Messages to communicate status or to provide instructions (whereas generated results are sent as Artifacts).
A Message can have multiple parts to denote different pieces of content. For example, a user request could include a textual description from a user and then multiple files used as context from the client.
The role of the message.
Type: Literal[‘user’, ‘agent’]
The parts of the message.
Type: list[Part]
Event type.
Type: Literal[‘message’]
Metadata about the message.
Type: NotRequired[dict[str, Any]]
Identifier created by the message creator.
Type: str
The context the message is associated with.
Type: NotRequired[str]
Identifier of task the message is related to.
Type: NotRequired[str]
Array of task IDs this message references.
Type: NotRequired[list[str]]
Array of extensions.
Type: NotRequired[list[str]]
Bases: _BasePart
A part that contains text.
The kind of the part.
Type: Literal[‘text’]
The text of the part.
Type: str
Bases: TypedDict
File with base64 encoded data.
The base64 encoded content of the file.
Type: str
Optional mime type for the file.
Type: NotRequired[str]
Bases: TypedDict
File with URI reference.
The URI of the file.
Type: str
The mime type of the file.
Type: NotRequired[str]
Bases: _BasePart
A part that contains a file.
The kind of the part.
Type: Literal[‘file’]
The file content - either bytes or URI.
Type: FileWithBytes | FileWithUri
Bases: _BasePart
A part that contains structured data.
The kind of the part.
Type: Literal[‘data’]
The data of the part.
Bases: TypedDict
Status and accompanying message for a task.
The current state of the task.
Type: TaskState
Additional status updates for client.
Type: NotRequired[Message]
ISO datetime value of when the status was updated.
Type: NotRequired[str]
Bases: TypedDict
A Task is a stateful entity that allows Clients and Remote Agents to achieve a specific outcome.
Clients and Remote Agents exchange Messages within a Task. Remote Agents generate results as Artifacts. A Task is always created by a Client and the status is always determined by the Remote Agent.
Unique identifier for the task.
Type: str
The context the task is associated with.
Type: str
Event type.
Type: Literal[‘task’]
Current status of the task.
Type: TaskStatus
Optional history of messages.
Type: NotRequired[list[Message]]
Collection of artifacts created by the agent.
Type: NotRequired[list[Artifact]]
Extension metadata.
Type: NotRequired[dict[str, Any]]
Bases: TypedDict
Sent by server during message/stream requests.
The id of the task.
Type: str
The context the task is associated with.
Type: str
Event type.
Type: Literal[‘status-update’]
The status of the task.
Type: TaskStatus
Indicates the end of the event stream.
Type: bool
Extension metadata.
Type: NotRequired[dict[str, Any]]
Bases: TypedDict
Sent by server during message/stream requests.
The id of the task.
Type: str
The context the task is associated with.
Type: str
Event type identification.
Type: Literal[‘artifact-update’]
The artifact that was updated.
Type: Artifact
Whether to append to existing artifact (true) or replace (false).
Type: NotRequired[bool]
Indicates this is the final chunk of the artifact.
Type: NotRequired[bool]
Extension metadata.
Type: NotRequired[dict[str, Any]]
Bases: TypedDict
Parameters for a task id.
The unique identifier for the task.
Type: str
Optional metadata associated with the request.
Type: NotRequired[dict[str, Any]]
Bases: TaskIdParams
Query parameters for a task.
Number of recent messages to be retrieved.
Type: NotRequired[int]
Bases: TypedDict
Configuration for the send message request.
Accepted output modalities by the client.
If the server should treat the client as a blocking request.
Type: NotRequired[bool]
Number of recent messages to be retrieved.
Type: NotRequired[int]
Where the server should send notifications when disconnected.
Type: NotRequired[PushNotificationConfig]
Bases: TypedDict
Parameters for message/send method.
Send message configuration.
Type: NotRequired[MessageSendConfiguration]
The message being sent to the server.
Type: Message
Extension metadata.
Type: NotRequired[dict[str, Any]]
Bases: TypedDict
Internal parameters for task execution within the framework.
Note: This is not part of the A2A protocol - it’s used internally for broker/worker communication.
The id of the task.
Type: str
The context id for the task.
Type: str
The message to process.
Type: Message
Number of recent messages to be retrieved.
Type: NotRequired[int]
Extension metadata.
Type: NotRequired[dict[str, Any]]
Bases: TypedDict
Parameters for getting list of pushNotificationConfigurations associated with a Task.
Task id.
Type: str
Extension metadata.
Type: NotRequired[dict[str, Any]]
Bases: TypedDict
Parameters for removing pushNotificationConfiguration associated with a Task.
Task id.
Type: str
The push notification config id to delete.
Type: str
Extension metadata.
Type: NotRequired[dict[str, Any]]
Bases: TypedDict
A JSON RPC message.
The JSON RPC version.
Type: Literal[‘2.0’]
The request id.
Bases: JSONRPCMessage, Generic[Method, Params]
A JSON RPC request.
The method to call.
Type: Method
The parameters to pass to the method.
Type: Params
Bases: TypedDict, Generic[CodeT, MessageT]
A JSON RPC error.
A number that indicates the error type that occurred.
Type: CodeT
A string providing a short description of the error.
Type: MessageT
A primitive or structured value containing additional information about the error.
Type: NotRequired[Any]
Bases: JSONRPCMessage, Generic[ResultT, ErrorT]
A JSON RPC response.
A security scheme for authentication.
Default: Annotated[Union[HttpSecurityScheme, ApiKeySecurityScheme, OAuth2SecurityScheme, OpenIdConnectSecurityScheme], pydantic.Field(discriminator='type')]
A fully formed piece of content exchanged between a client and a remote agent as part of a Message or an Artifact.
Each Part has its own content type and metadata.
Default: Annotated[Union[TextPart, FilePart, DataPart], pydantic.Field(discriminator='kind')]
The possible states of a task.
Type: TypeAlias Default: Literal['submitted', 'working', 'input-required', 'completed', 'canceled', 'failed', 'rejected', 'auth-required', 'unknown']
A JSON RPC error for a parse error.
Default: JSONRPCError[Literal[-32700], Literal['Invalid JSON payload']]
A JSON RPC error for an invalid request.
Default: JSONRPCError[Literal[-32600], Literal['Request payload validation error']]
A JSON RPC error for a method not found.
Default: JSONRPCError[Literal[-32601], Literal['Method not found']]
A JSON RPC error for invalid parameters.
Default: JSONRPCError[Literal[-32602], Literal['Invalid parameters']]
A JSON RPC error for an internal error.
Default: JSONRPCError[Literal[-32603], Literal['Internal error']]
A JSON RPC error for a task not found.
Default: JSONRPCError[Literal[-32001], Literal['Task not found']]
A JSON RPC error for a task not cancelable.
Default: JSONRPCError[Literal[-32002], Literal['Task not cancelable']]
A JSON RPC error for a push notification not supported.
Default: JSONRPCError[Literal[-32003], Literal['Push notification not supported']]
A JSON RPC error for an unsupported operation.
Default: JSONRPCError[Literal[-32004], Literal['This operation is not supported']]
A JSON RPC error for incompatible content types.
Default: JSONRPCError[Literal[-32005], Literal['Incompatible content types']]
A JSON RPC error for invalid agent response.
Default: JSONRPCError[Literal[-32006], Literal['Invalid agent response']]
A JSON RPC request to send a message.
Default: JSONRPCRequest[Literal['message/send'], MessageSendParams]
A JSON RPC response to send a message.
Default: JSONRPCResponse[Union[Task, Message], JSONRPCError[Any, Any]]
A JSON RPC request to stream a message.
Default: JSONRPCRequest[Literal['message/stream'], MessageSendParams]
A JSON RPC response to a StreamMessageRequest.
Default: JSONRPCResponse[Union[Task, Message, TaskStatusUpdateEvent, TaskArtifactUpdateEvent], JSONRPCError[Any, Any]]
A JSON RPC request to get a task.
Default: JSONRPCRequest[Literal['tasks/get'], TaskQueryParams]
A JSON RPC response to get a task.
Default: JSONRPCResponse[Task, TaskNotFoundError]
A JSON RPC request to cancel a task.
Default: JSONRPCRequest[Literal['tasks/cancel'], TaskIdParams]
A JSON RPC response to cancel a task.
Default: JSONRPCResponse[Task, Union[TaskNotCancelableError, TaskNotFoundError]]
A JSON RPC request to set a task push notification.
Default: JSONRPCRequest[Literal['tasks/pushNotification/set'], TaskPushNotificationConfig]
A JSON RPC response to set a task push notification.
Default: JSONRPCResponse[TaskPushNotificationConfig, PushNotificationNotSupportedError]
A JSON RPC request to get a task push notification.
Default: JSONRPCRequest[Literal['tasks/pushNotification/get'], TaskIdParams]
A JSON RPC response to get a task push notification.
Default: JSONRPCResponse[TaskPushNotificationConfig, PushNotificationNotSupportedError]
A JSON RPC request to resubscribe to a task.
Default: JSONRPCRequest[Literal['tasks/resubscribe'], TaskIdParams]
A JSON RPC request to list task push notification configs.
Default: JSONRPCRequest[Literal['tasks/pushNotificationConfig/list'], ListTaskPushNotificationConfigParams]
A JSON RPC request to delete a task push notification config.
Default: JSONRPCRequest[Literal['tasks/pushNotificationConfig/delete'], DeleteTaskPushNotificationConfigParams]
A JSON RPC request to the A2A server.
Default: Annotated[Union[SendMessageRequest, StreamMessageRequest, GetTaskRequest, CancelTaskRequest, SetTaskPushNotificationRequest, GetTaskPushNotificationRequest, ResubscribeTaskRequest, ListTaskPushNotificationConfigRequest, DeleteTaskPushNotificationConfigRequest], Discriminator('method')]
A JSON RPC response from the A2A server.
Type: TypeAlias Default: Union[SendMessageResponse, StreamMessageResponse, GetTaskResponse, CancelTaskResponse, SetTaskPushNotificationResponse, GetTaskPushNotificationResponse]
A client for the A2A protocol.
@async
def send_message(
message: Message,
metadata: dict[str, Any] | None = None,
configuration: MessageSendConfiguration | None = None,
) -> SendMessageResponse
Send a message using the A2A protocol.
Returns a JSON-RPC response containing either a result (Task) or an error.
SendMessageResponse
Bases: Exception
An error raised when an unexpected response is received from the server.