fasta2a

FastA2A

基类：Starlette

FastA2A库的主类。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/applications.py

class FastA2A(Starlette):
    """The main class for the FastA2A library."""

    def __init__(
        self,
        *,
        storage: Storage,
        broker: Broker,
        # Agent card
        name: str | None = None,
        url: str = 'https://:8000',
        version: str = '1.0.0',
        description: str | None = None,
        provider: AgentProvider | None = None,
        skills: list[Skill] | None = None,
        # Starlette
        debug: bool = False,
        routes: Sequence[Route] | None = None,
        middleware: Sequence[Middleware] | None = None,
        exception_handlers: dict[Any, ExceptionHandler] | None = None,
        lifespan: Lifespan[FastA2A] | None = None,
    ):
        if lifespan is None:
            lifespan = _default_lifespan

        super().__init__(
            debug=debug,
            routes=routes,
            middleware=middleware,
            exception_handlers=exception_handlers,
            lifespan=lifespan,
        )

        self.name = name or 'My Agent'
        self.url = url
        self.version = version
        self.description = description
        self.provider = provider
        self.skills = skills or []
        # NOTE: For now, I don't think there's any reason to support any other input/output modes.
        self.default_input_modes = ['application/json']
        self.default_output_modes = ['application/json']

        self.task_manager = TaskManager(broker=broker, storage=storage)

        # Setup
        self._agent_card_json_schema: bytes | None = None
        self.router.add_route('/.well-known/agent.json', self._agent_card_endpoint, methods=['HEAD', 'GET', 'OPTIONS'])
        self.router.add_route('/', self._agent_run_endpoint, methods=['POST'])
        self.router.add_route('/docs', self._docs_endpoint, methods=['GET'])

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope['type'] == 'http' and not self.task_manager.is_running:
            raise RuntimeError('TaskManager was not properly initialized.')
        await super().__call__(scope, receive, send)

    async def _agent_card_endpoint(self, request: Request) -> Response:
        if self._agent_card_json_schema is None:
            agent_card = AgentCard(
                name=self.name,
                description=self.description or 'An AI agent exposed as an A2A agent.',
                url=self.url,
                version=self.version,
                protocol_version='0.2.5',
                skills=self.skills,
                default_input_modes=self.default_input_modes,
                default_output_modes=self.default_output_modes,
                capabilities=AgentCapabilities(
                    streaming=False, push_notifications=False, state_transition_history=False
                ),
            )
            if self.provider is not None:
                agent_card['provider'] = self.provider
            self._agent_card_json_schema = agent_card_ta.dump_json(agent_card, by_alias=True)
        return Response(content=self._agent_card_json_schema, media_type='application/json')

    async def _docs_endpoint(self, request: Request) -> Response:
        """Serve the documentation interface."""
        docs_path = Path(__file__).parent / 'static' / 'docs.html'
        return FileResponse(docs_path, media_type='text/html')

    async def _agent_run_endpoint(self, request: Request) -> Response:
        """This is the main endpoint for the A2A server.

        Although the specification allows freedom of choice and implementation, I'm pretty sure about some decisions.

        1. The server will always either send a "submitted" or a "failed" on `tasks/send`.
            Never a "completed" on the first message.
        2. There are three possible ends for the task:
            2.1. The task was "completed" successfully.
            2.2. The task was "canceled".
            2.3. The task "failed".
        3. The server will send a "working" on the first chunk on `tasks/pushNotification/get`.
        """
        data = await request.body()
        a2a_request = a2a_request_ta.validate_json(data)

        if a2a_request['method'] == 'message/send':
            jsonrpc_response = await self.task_manager.send_message(a2a_request)
        elif a2a_request['method'] == 'tasks/get':
            jsonrpc_response = await self.task_manager.get_task(a2a_request)
        elif a2a_request['method'] == 'tasks/cancel':
            jsonrpc_response = await self.task_manager.cancel_task(a2a_request)
        else:
            raise NotImplementedError(f'Method {a2a_request["method"]} not implemented.')
        return Response(
            content=a2a_response_ta.dump_json(jsonrpc_response, by_alias=True), media_type='application/json'
        )

代理（Broker） 数据类

基类：ABC

代理类负责调度任务。

HTTP服务器使用代理来调度任务。

简单的实现是InMemoryBroker，它在与HTTP服务器相同的进程中运行任务。尽管如此，这个类可以被扩展以支持远程工作单元。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/broker.py

@dataclass
class Broker(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(self, params: TaskSendParams) -> None:
        """Send a task to be executed by the worker."""
        raise NotImplementedError('send_run_task is not implemented yet.')

    @abstractmethod
    async def cancel_task(self, params: TaskIdParams) -> None:
        """Cancel a task."""
        raise NotImplementedError('send_cancel_task is not implemented yet.')

    @abstractmethod
    async def __aenter__(self) -> Self: ...

    @abstractmethod
    async def __aexit__(self, exc_type: Any, exc_value: Any, traceback: Any): ...

    @abstractmethod
    def receive_task_operations(self) -> 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.
        """

运行任务 抽象方法 异步

run_task(params: TaskSendParams) -> None

发送一个任务给工作单元执行。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/broker.py

@abstractmethod
async def run_task(self, params: TaskSendParams) -> None:
    """Send a task to be executed by the worker."""
    raise NotImplementedError('send_run_task is not implemented yet.')

取消任务 抽象方法 异步

cancel_task(params: TaskIdParams) -> None

取消一个任务。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/broker.py

@abstractmethod
async def cancel_task(self, params: TaskIdParams) -> None:
    """Cancel a task."""
    raise NotImplementedError('send_cancel_task is not implemented yet.')

接收任务操作 抽象方法

receive_task_operations() -> AsyncIterator[TaskOperation]

从代理接收任务操作。

在多工作单元的设置中，代理需要在工作单元之间轮询任务操作。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/broker.py

@abstractmethod
def receive_task_operations(self) -> 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.
    """

技能

基类：TypedDict

技能是智能体可以执行的能力单元。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class Skill(TypedDict):
    """Skills are a unit of capability that an agent can perform."""

    id: str
    """A unique identifier for the skill."""

    name: str
    """Human readable name of the skill."""

    description: 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.
    """

    tags: list[str]
    """Set of tag-words describing classes of capabilities for this specific skill.

    Examples: "cooking", "customer support", "billing".
    """

    examples: NotRequired[list[str]]
    """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")
    """

    input_modes: list[str]
    """Supported mime types for input data."""

    output_modes: list[str]
    """Supported mime types for output data."""

id 实例属性

id: str

技能的唯一标识符。

name instance-attribute

name: str

技能的人类可读名称。

description 实例属性

description: str

技能的人类可读描述。

它将被客户端或人类用作理解技能的提示。

tags 实例属性

tags: list[str]

一组描述此特定技能能力类别的标签词。

示例：“烹饪”、“客户支持”、“计费”。

examples 实例属性

examples: NotRequired[list[str]]

该技能可以执行的示例场景集合。

将被客户端用作理解如何使用该技能的提示。（例如：“我需要一个面包的食谱”）

input_modes 实例属性

input_modes: list[str]

支持的输入数据MIME类型。

output_modes 实例属性

output_modes: list[str]

支持的输出数据MIME类型。

存储

基类：ABC, Generic[ContextT]

用于检索和保存任务，以及检索和保存上下文的存储。

存储有两个目的：1. 任务存储：以A2A协议格式存储任务及其状态、产物和消息历史。2. 上下文存储：以针对特定智能体实现优化的格式存储对话上下文。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/storage.py

class Storage(ABC, Generic[ContextT]):
    """A storage to retrieve and save tasks, as well as retrieve and save context.

    The storage serves two purposes:
    1. Task storage: Stores tasks in A2A protocol format with their status, artifacts, and message history
    2. Context storage: Stores conversation context in a format optimized for the specific agent implementation
    """

    @abstractmethod
    async def load_task(self, task_id: str, history_length: int | None = None) -> Task | None:
        """Load a task from storage.

        If the task is not found, return None.
        """

    @abstractmethod
    async def submit_task(self, context_id: str, message: Message) -> Task:
        """Submit a task to storage."""

    @abstractmethod
    async def update_task(
        self,
        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."""

    @abstractmethod
    async def load_context(self, context_id: str) -> ContextT | None:
        """Retrieve the stored context given the `context_id`."""

    @abstractmethod
    async def update_context(self, context_id: str, context: ContextT) -> None:
        """Updates the context for a `context_id`.

        Implementing agent can decide what to store in context.
        """

加载任务 抽象方法 异步

load_task(
    task_id: str, history_length: int | None = None
) -> Task | None

从存储中加载一个任务。

如果未找到任务，则返回None。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/storage.py

@abstractmethod
async def load_task(self, task_id: str, history_length: int | None = None) -> Task | None:
    """Load a task from storage.

    If the task is not found, return None.
    """

提交任务 抽象方法 异步

submit_task(context_id: str, message: Message) -> Task

向存储提交一个任务。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/storage.py

@abstractmethod
async def submit_task(self, context_id: str, message: Message) -> Task:
    """Submit a task to storage."""

更新任务 抽象方法 异步

update_task(
    task_id: str,
    state: TaskState,
    new_artifacts: list[Artifact] | None = None,
    new_messages: list[Message] | None = None,
) -> Task

更新任务的状态。如果指定，则追加产物和消息。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/storage.py

@abstractmethod
async def update_task(
    self,
    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."""

加载上下文 抽象方法 异步

load_context(context_id: str) -> ContextT | None

根据context_id检索存储的上下文。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/storage.py

@abstractmethod
async def load_context(self, context_id: str) -> ContextT | None:
    """Retrieve the stored context given the `context_id`."""

更新上下文 抽象方法 异步

update_context(context_id: str, context: ContextT) -> None

更新context_id的上下文。

实现智能体可以决定在上下文中存储什么。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/storage.py

@abstractmethod
async def update_context(self, context_id: str, context: ContextT) -> None:
    """Updates the context for a `context_id`.

    Implementing agent can decide what to store in context.
    """

工作单元（Worker） 数据类

基类：ABC, Generic[ContextT]

工作单元负责执行任务。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/worker.py

@dataclass
class Worker(ABC, Generic[ContextT]):
    """A worker is responsible for executing tasks."""

    broker: Broker
    storage: Storage[ContextT]

    @asynccontextmanager
    async def run(self) -> AsyncIterator[None]:
        """Run the worker.

        It connects to the broker, and it makes itself available to receive commands.
        """
        async with anyio.create_task_group() as tg:
            tg.start_soon(self._loop)
            yield
            tg.cancel_scope.cancel()

    async def _loop(self) -> None:
        async for task_operation in self.broker.receive_task_operations():
            await self._handle_task_operation(task_operation)

    async def _handle_task_operation(self, task_operation: TaskOperation) -> None:
        try:
            with use_span(task_operation['_current_span']):
                with tracer.start_as_current_span(
                    f'{task_operation["operation"]} task', attributes={'logfire.tags': ['fasta2a']}
                ):
                    if task_operation['operation'] == 'run':
                        await self.run_task(task_operation['params'])
                    elif task_operation['operation'] == 'cancel':
                        await self.cancel_task(task_operation['params'])
                    else:
                        assert_never(task_operation)
        except Exception:
            await self.storage.update_task(task_operation['params']['id'], state='failed')

    @abstractmethod
    async def run_task(self, params: TaskSendParams) -> None: ...

    @abstractmethod
    async def cancel_task(self, params: TaskIdParams) -> None: ...

    @abstractmethod
    def build_message_history(self, history: list[Message]) -> list[Any]: ...

    @abstractmethod
    def build_artifacts(self, result: Any) -> list[Artifact]: ...

run async

run() -> AsyncIterator[None]

运行工作单元。

它连接到代理，并使自己可用于接收命令。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/worker.py

@asynccontextmanager
async def run(self) -> AsyncIterator[None]:
    """Run the worker.

    It connects to the broker, and it makes itself available to receive commands.
    """
    async with anyio.create_task_group() as tg:
        tg.start_soon(self._loop)
        yield
        tg.cancel_scope.cancel()

此模块包含智能体卡片的模式。

智能体卡片（AgentCard）

基类：TypedDict

描述一个智能体的卡片。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class AgentCard(TypedDict):
    """The card that describes an agent."""

    name: str
    """Human readable name of the agent e.g. "Recipe Agent"."""

    description: 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.")
    """

    url: str
    """A URL to the address the agent is hosted at."""

    version: str
    """The version of the agent - format is up to the provider. (e.g. "1.0.0")"""

    protocol_version: str
    """The version of the A2A protocol this agent supports."""

    provider: NotRequired[AgentProvider]
    """The service provider of the agent."""

    documentation_url: NotRequired[str]
    """A URL to documentation for the agent."""

    icon_url: NotRequired[str]
    """A URL to an icon for the agent."""

    preferred_transport: NotRequired[str]
    """The transport of the preferred endpoint. If empty, defaults to JSONRPC."""

    additional_interfaces: NotRequired[list[AgentInterface]]
    """Announcement of additional supported transports."""

    capabilities: AgentCapabilities
    """The capabilities of the agent."""

    security: NotRequired[list[dict[str, list[str]]]]
    """Security requirements for contacting the agent."""

    security_schemes: NotRequired[dict[str, SecurityScheme]]
    """Security scheme definitions."""

    default_input_modes: list[str]
    """Supported mime types for input data."""

    default_output_modes: list[str]
    """Supported mime types for output data."""

    skills: list[Skill]

name instance-attribute

name: str

智能体的人类可读名称，例如“食谱智能体”。

description 实例属性

description: str

智能体的人类可读描述。

用于帮助用户和其他智能体理解该智能体能做什么。（例如“帮助用户处理食谱和烹饪的智能体。”）

url 实例属性

url: str

智能体托管地址的URL。

version 实例属性

version: str

智能体的版本 - 格式由提供商决定。（例如“1.0.0”）

protocol_version 实例属性

protocol_version: str

此智能体支持的A2A协议版本。

provider 实例属性

provider: NotRequired[AgentProvider]

智能体的服务提供商。

documentation_url 实例属性

documentation_url: NotRequired[str]

智能体文档的URL。

icon_url 实例属性

icon_url: NotRequired[str]

智能体图标的URL。

preferred_transport 实例属性

preferred_transport: NotRequired[str]

首选端点的传输方式。如果为空，则默认为JSONRPC。

additional_interfaces 实例属性

additional_interfaces: NotRequired[list[AgentInterface]]

对其他支持的传输方式的声明。

capabilities 实例属性

capabilities: AgentCapabilities

智能体的能力。

security 实例属性

security: NotRequired[list[dict[str, list[str]]]]

联系智能体的安全要求。

security_schemes 实例属性

security_schemes: NotRequired[dict[str, SecurityScheme]]

安全方案定义。

default_input_modes 实例属性

default_input_modes: list[str]

支持的输入数据MIME类型。

default_output_modes 实例属性

default_output_modes: list[str]

支持的输出数据MIME类型。

智能体提供商

基类：TypedDict

智能体的服务提供商。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

class AgentProvider(TypedDict):
    """The service provider of the agent."""

    organization: str
    url: str

智能体能力

基类：TypedDict

智能体的能力。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class AgentCapabilities(TypedDict):
    """The capabilities of the agent."""

    streaming: NotRequired[bool]
    """Whether the agent supports streaming."""

    push_notifications: NotRequired[bool]
    """Whether the agent can notify updates to client."""

    state_transition_history: NotRequired[bool]
    """Whether the agent exposes status change history for tasks."""

streaming 实例属性

streaming: NotRequired[bool]

智能体是否支持流式传输。

push_notifications 实例属性

push_notifications: NotRequired[bool]

智能体是否可以向客户端通知更新。

state_transition_history 实例属性

state_transition_history: NotRequired[bool]

智能体是否暴露任务的状态变更历史。

HTTP安全方案

基类：TypedDict

HTTP安全方案。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class HttpSecurityScheme(TypedDict):
    """HTTP security scheme."""

    type: Literal['http']
    scheme: str
    """The name of the HTTP Authorization scheme."""
    bearer_format: NotRequired[str]
    """A hint to the client to identify how the bearer token is formatted."""
    description: NotRequired[str]
    """Description of this security scheme."""

scheme 实例属性

scheme: str

HTTP授权方案的名称。

bearer_format 实例属性

bearer_format: NotRequired[str]

给客户端的提示，用于识别持有者令牌的格式。

description 实例属性

description: NotRequired[str]

此安全方案的描述。

API密钥安全方案

基类：TypedDict

API密钥安全方案。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class ApiKeySecurityScheme(TypedDict):
    """API Key security scheme."""

    type: Literal['apiKey']
    name: str
    """The name of the header, query or cookie parameter to be used."""
    in_: Literal['query', 'header', 'cookie']
    """The location of the API key."""
    description: NotRequired[str]
    """Description of this security scheme."""

name instance-attribute

name: str

要使用的头、查询或cookie参数的名称。

in_ 实例属性

in_: Literal['query', 'header', 'cookie']

API密钥的位置。

description 实例属性

description: NotRequired[str]

此安全方案的描述。

OAuth2安全方案

基类：TypedDict

OAuth2安全方案。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class OAuth2SecurityScheme(TypedDict):
    """OAuth2 security scheme."""

    type: Literal['oauth2']
    flows: dict[str, Any]
    """An object containing configuration information for the flow types supported."""
    description: NotRequired[str]
    """Description of this security scheme."""

flows 实例属性

flows: dict[str, Any]

一个包含支持的流程类型配置信息的对象。

description 实例属性

description: NotRequired[str]

此安全方案的描述。

OpenID Connect安全方案

基类：TypedDict

OpenID Connect安全方案。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class OpenIdConnectSecurityScheme(TypedDict):
    """OpenID Connect security scheme."""

    type: Literal['openIdConnect']
    open_id_connect_url: str
    """OpenId Connect URL to discover OAuth2 configuration values."""
    description: NotRequired[str]
    """Description of this security scheme."""

open_id_connect_url 实例属性

open_id_connect_url: str

用于发现OAuth2配置值的OpenID Connect URL。

description 实例属性

description: NotRequired[str]

此安全方案的描述。

安全方案 模块属性

SecurityScheme = Annotated[
    Union[
        HttpSecurityScheme,
        ApiKeySecurityScheme,
        OAuth2SecurityScheme,
        OpenIdConnectSecurityScheme,
    ],
    Field(discriminator="type"),
]

用于身份验证的安全方案。

智能体接口

基类：TypedDict

智能体支持的接口。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class AgentInterface(TypedDict):
    """An interface that the agent supports."""

    transport: str
    """The transport protocol (e.g., 'jsonrpc', 'websocket')."""

    url: str
    """The URL endpoint for this transport."""

    description: NotRequired[str]
    """Description of this interface."""

transport 实例属性

transport: str

传输协议（例如，“jsonrpc”、“websocket”）。

url 实例属性

url: str

此传输的URL端点。

description 实例属性

description: NotRequired[str]

此接口的描述。

智能体扩展

基类：TypedDict

智能体支持的扩展声明。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class AgentExtension(TypedDict):
    """A declaration of an extension supported by an Agent."""

    uri: str
    """The URI of the extension."""

    description: NotRequired[str]
    """A description of how this agent uses this extension."""

    required: NotRequired[bool]
    """Whether the client must follow specific requirements of the extension."""

    params: NotRequired[dict[str, Any]]
    """Optional configuration for the extension."""

uri 实例属性

uri: str

扩展的URI。

description 实例属性

description: NotRequired[str]

关于此智能体如何使用此扩展的描述。

required 实例属性

required: NotRequired[bool]

客户端是否必须遵循扩展的特定要求。

params 实例属性

params: NotRequired[dict[str, Any]]

扩展的可选配置。

技能

基类：TypedDict

技能是智能体可以执行的能力单元。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class Skill(TypedDict):
    """Skills are a unit of capability that an agent can perform."""

    id: str
    """A unique identifier for the skill."""

    name: str
    """Human readable name of the skill."""

    description: 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.
    """

    tags: list[str]
    """Set of tag-words describing classes of capabilities for this specific skill.

    Examples: "cooking", "customer support", "billing".
    """

    examples: NotRequired[list[str]]
    """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")
    """

    input_modes: list[str]
    """Supported mime types for input data."""

    output_modes: list[str]
    """Supported mime types for output data."""

id 实例属性

id: str

技能的唯一标识符。

name instance-attribute

name: str

技能的人类可读名称。

description 实例属性

description: str

技能的人类可读描述。

它将被客户端或人类用作理解技能的提示。

tags 实例属性

tags: list[str]

一组描述此特定技能能力类别的标签词。

示例：“烹饪”、“客户支持”、“计费”。

examples 实例属性

examples: NotRequired[list[str]]

该技能可以执行的示例场景集合。

将被客户端用作理解如何使用该技能的提示。（例如：“我需要一个面包的食谱”）

input_modes 实例属性

input_modes: list[str]

支持的输入数据MIME类型。

output_modes 实例属性

output_modes: list[str]

支持的输出数据MIME类型。

产物（Artifact）

基类：TypedDict

智能体生成产物作为任务的最终结果。

产物是不可变的，可以命名，并且可以有多个部分。流式响应可以向现有产物追加部分。

单个任务可以生成多个产物。例如，“创建一个网页”可以创建单独的HTML和图像产物。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class Artifact(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.
    """

    artifact_id: str
    """Unique identifier for the artifact."""

    name: NotRequired[str]
    """The name of the artifact."""

    description: NotRequired[str]
    """A description of the artifact."""

    parts: list[Part]
    """The parts that make up the artifact."""

    metadata: NotRequired[dict[str, Any]]
    """Metadata about the artifact."""

    extensions: NotRequired[list[str]]
    """Array of extensions."""

    append: NotRequired[bool]
    """Whether to append this artifact to an existing one."""

    last_chunk: NotRequired[bool]
    """Whether this is the last chunk of the artifact."""

artifact_id 实例属性

artifact_id: str

产物的唯一标识符。

name instance-attribute

name: NotRequired[str]

产物的名称。

description 实例属性

description: NotRequired[str]

产物的描述。

parts 实例属性

parts: list[Part]

构成产物的各个部分。

metadata 实例属性

metadata: NotRequired[dict[str, Any]]

关于产物的元数据。

extensions 实例属性

extensions: NotRequired[list[str]]

扩展数组。

append 实例属性

append: NotRequired[bool]

是否将此产物追加到现有产物上。

last_chunk 实例属性

last_chunk: NotRequired[bool]

这是否是产物的最后一块。

推送通知配置

基类：TypedDict

推送通知的配置。

A2A支持一种安全的通知机制，智能体可以通过推送通知服务在连接会话之外通知客户端更新。在企业内部和跨企业环境中，智能体验证通知服务的身份、向服务进行身份验证并提供一个将通知与执行任务关联起来的标识符至关重要。

推送通知服务的目标服务器应被视为一个独立的服务，并且不保证（甚至不期望）直接是客户端。此推送通知服务负责对智能体进行身份验证和授权，并将经过验证的通知代理到适当的端点（这可以是任何东西，从发布/订阅队列到电子邮件收件箱或其他服务等）。

对于独立的客户端-智能体对（例如，在受限VPC中的本地服务网格等）或没有企业安全问题的隔离环境中的特定场景，客户端可以选择简单地打开一个端口并充当自己的推送通知服务。任何企业级实现都可能有一个集中的服务，用受信任的通知凭据对远程智能体进行身份验证，并能处理在线/离线场景。（这应被视为类似于移动推送通知服务）。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class PushNotificationConfig(TypedDict):
    """Configuration for push notifications.

    A2A supports a secure notification mechanism whereby an agent can notify a client of an update
    outside of 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).
    """

    id: NotRequired[str]
    """Server-assigned identifier."""

    url: str
    """The URL to send push notifications to."""

    token: NotRequired[str]
    """Token unique to this task/session."""

    authentication: NotRequired[SecurityScheme]
    """Authentication details for push notifications."""

id 实例属性

id: NotRequired[str]

服务器分配的标识符。

url 实例属性

url: str

发送推送通知的URL。

token 实例属性

token: NotRequired[str]

此任务/会话唯一的令牌。

authentication 实例属性

authentication: NotRequired[SecurityScheme]

推送通知的认证详情。

任务推送通知配置

基类：TypedDict

任务推送通知的配置。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class TaskPushNotificationConfig(TypedDict):
    """Configuration for task push notifications."""

    id: str
    """The task id."""

    push_notification_config: PushNotificationConfig
    """The push notification configuration."""

id 实例属性

id: str

任务ID。

push_notification_config 实例属性

push_notification_config: PushNotificationConfig

推送通知配置。

消息

基类：TypedDict

消息包含任何非产物的内容。

这可以包括智能体的思考、用户上下文、指令、错误、状态或元数据等。

所有来自客户端的内容都以消息的形式出现。智能体发送消息来沟通状态或提供指令（而生成的结果则作为产物发送）。

一条消息可以有多个部分来表示不同的内容片段。例如，一个用户请求可能包括用户的文本描述，然后是客户端用作上下文的多个文件。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class Message(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.
    """

    role: Literal['user', 'agent']
    """The role of the message."""

    parts: list[Part]
    """The parts of the message."""

    kind: Literal['message']
    """Event type."""

    metadata: NotRequired[dict[str, Any]]
    """Metadata about the message."""

    # Additional fields
    message_id: str
    """Identifier created by the message creator."""

    context_id: NotRequired[str]
    """The context the message is associated with."""

    task_id: NotRequired[str]
    """Identifier of task the message is related to."""

    reference_task_ids: NotRequired[list[str]]
    """Array of task IDs this message references."""

    extensions: NotRequired[list[str]]
    """Array of extensions."""

role 实例属性

role: Literal['user', 'agent']

消息的角色。

parts 实例属性

parts: list[Part]

消息的各个部分。

kind 实例属性

kind: Literal['message']

事件类型。

metadata 实例属性

metadata: NotRequired[dict[str, Any]]

关于消息的元数据。

message_id 实例属性

message_id: str

由消息创建者创建的标识符。

context_id 实例属性

context_id: NotRequired[str]

消息关联的上下文。

task_id 实例属性

task_id: NotRequired[str]

消息相关的任务标识符。

reference_task_ids 实例属性

reference_task_ids: NotRequired[list[str]]

此消息引用的任务ID数组。

extensions 实例属性

extensions: NotRequired[list[str]]

扩展数组。

文本部分

基类：_BasePart

包含文本的部分。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class TextPart(_BasePart):
    """A part that contains text."""

    kind: Literal['text']
    """The kind of the part."""

    text: str
    """The text of the part."""

kind 实例属性

kind: Literal['text']

部分的种类。

text 实例属性

text: str

部分的文本。

带字节的文件

基类：TypedDict

带有base64编码数据的文件。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class FileWithBytes(TypedDict):
    """File with base64 encoded data."""

    bytes: str
    """The base64 encoded content of the file."""

    mime_type: NotRequired[str]
    """Optional mime type for the file."""

bytes 实例属性

bytes: str

文件的base64编码内容。

mime_type 实例属性

mime_type: NotRequired[str]

文件的可选MIME类型。

带URI的文件

基类：TypedDict

带有URI引用的文件。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class FileWithUri(TypedDict):
    """File with URI reference."""

    uri: str
    """The URI of the file."""

    mime_type: NotRequired[str]
    """The mime type of the file."""

uri 实例属性

uri: str

文件的URI。

mime_type 实例属性

mime_type: NotRequired[str]

文件的MIME类型。

文件部分

基类：_BasePart

包含文件的部分。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class FilePart(_BasePart):
    """A part that contains a file."""

    kind: Literal['file']
    """The kind of the part."""

    file: FileWithBytes | FileWithUri
    """The file content - either bytes or URI."""

kind 实例属性

kind: Literal['file']

部分的种类。

file 实例属性

file: FileWithBytes | FileWithUri

文件内容 - 字节或URI。

数据部分

基类：_BasePart

包含结构化数据的部分。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class DataPart(_BasePart):
    """A part that contains structured data."""

    kind: Literal['data']
    """The kind of the part."""

    data: dict[str, Any]
    """The data of the part."""

kind 实例属性

kind: Literal['data']

部分的种类。

data 实例属性

data: dict[str, Any]

部分的数据。

Part 模块属性

Part = Annotated[
    Union[TextPart, FilePart, DataPart],
    Field(discriminator="kind"),
]

作为消息或产物的一部分，在客户端和远程智能体之间交换的完整内容片段。

每个部分都有自己的内容类型和元数据。

TaskState 模块属性

TaskState: TypeAlias = Literal[
    "submitted",
    "working",
    "input-required",
    "completed",
    "canceled",
    "failed",
    "rejected",
    "auth-required",
    "unknown",
]

任务的可能状态。

任务状况

基类：TypedDict

任务的状态及附带消息。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class TaskStatus(TypedDict):
    """Status and accompanying message for a task."""

    state: TaskState
    """The current state of the task."""

    message: NotRequired[Message]
    """Additional status updates for client."""

    timestamp: NotRequired[str]
    """ISO datetime value of when the status was updated."""

state 实例属性

state: TaskState

任务的当前状态。

message 实例属性

message: NotRequired[Message]

给客户端的额外状态更新。

timestamp 实例属性

timestamp: NotRequired[str]

状态更新时的ISO日期时间值。

任务

基类：TypedDict

任务是一个有状态的实体，允许客户端和远程智能体实现特定的结果。

客户端和远程智能体在任务内交换消息。远程智能体生成结果作为产物。任务总是由客户端创建，状态总是由远程智能体确定。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class Task(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.
    """

    id: str
    """Unique identifier for the task."""

    context_id: str
    """The context the task is associated with."""

    kind: Literal['task']
    """Event type."""

    status: TaskStatus
    """Current status of the task."""

    history: NotRequired[list[Message]]
    """Optional history of messages."""

    artifacts: NotRequired[list[Artifact]]
    """Collection of artifacts created by the agent."""

    metadata: NotRequired[dict[str, Any]]
    """Extension metadata."""

id 实例属性

id: str

任务的唯一标识符。

context_id 实例属性

context_id: str

任务关联的上下文。

kind 实例属性

kind: Literal['task']

事件类型。

status 实例属性

status: TaskStatus

任务的当前状况。

history 实例属性

history: NotRequired[list[Message]]

可选的消息历史。

artifacts 实例属性

artifacts: NotRequired[list[Artifact]]

由智能体创建的产物集合。

metadata 实例属性

metadata: NotRequired[dict[str, Any]]

扩展元数据。

任务状况更新事件

基类：TypedDict

在消息/流请求期间由服务器发送。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class TaskStatusUpdateEvent(TypedDict):
    """Sent by server during message/stream requests."""

    task_id: str
    """The id of the task."""

    context_id: str
    """The context the task is associated with."""

    kind: Literal['status-update']
    """Event type."""

    status: TaskStatus
    """The status of the task."""

    final: bool
    """Indicates the end of the event stream."""

    metadata: NotRequired[dict[str, Any]]
    """Extension metadata."""

task_id 实例属性

task_id: str

任务的ID。

context_id 实例属性

context_id: str

任务关联的上下文。

kind 实例属性

kind: Literal['status-update']

事件类型。

status 实例属性

status: TaskStatus

任务的状况。

final 实例属性

final: bool

表示事件流的结束。

metadata 实例属性

metadata: NotRequired[dict[str, Any]]

扩展元数据。

任务产物更新事件

基类：TypedDict

在消息/流请求期间由服务器发送。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class TaskArtifactUpdateEvent(TypedDict):
    """Sent by server during message/stream requests."""

    task_id: str
    """The id of the task."""

    context_id: str
    """The context the task is associated with."""

    kind: Literal['artifact-update']
    """Event type identification."""

    artifact: Artifact
    """The artifact that was updated."""

    append: NotRequired[bool]
    """Whether to append to existing artifact (true) or replace (false)."""

    last_chunk: NotRequired[bool]
    """Indicates this is the final chunk of the artifact."""

    metadata: NotRequired[dict[str, Any]]
    """Extension metadata."""

task_id 实例属性

task_id: str

任务的ID。

context_id 实例属性

context_id: str

任务关联的上下文。

kind 实例属性

kind: Literal['artifact-update']

事件类型标识。

artifact 实例属性

artifact: Artifact

已更新的产物。

append 实例属性

append: NotRequired[bool]

是追加到现有产物（true）还是替换（false）。

last_chunk 实例属性

last_chunk: NotRequired[bool]

表示这是产物的最后一块。

metadata 实例属性

metadata: NotRequired[dict[str, Any]]

扩展元数据。

任务ID参数

基类：TypedDict

任务ID的参数。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class TaskIdParams(TypedDict):
    """Parameters for a task id."""

    id: str
    metadata: NotRequired[dict[str, Any]]

任务查询参数

基类：TaskIdParams

任务的查询参数。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class TaskQueryParams(TaskIdParams):
    """Query parameters for a task."""

    history_length: NotRequired[int]
    """Number of recent messages to be retrieved."""

history_length 实例属性

history_length: NotRequired[int]

要检索的最近消息数量。

消息发送配置

基类：TypedDict

发送消息请求的配置。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class MessageSendConfiguration(TypedDict):
    """Configuration for the send message request."""

    accepted_output_modes: list[str]
    """Accepted output modalities by the client."""

    blocking: NotRequired[bool]
    """If the server should treat the client as a blocking request."""

    history_length: NotRequired[int]
    """Number of recent messages to be retrieved."""

    push_notification_config: NotRequired[PushNotificationConfig]
    """Where the server should send notifications when disconnected."""

accepted_output_modes 实例属性

accepted_output_modes: list[str]

客户端接受的输出模式。

blocking 实例属性

blocking: NotRequired[bool]

服务器是否应将客户端视为阻塞请求。

history_length 实例属性

history_length: NotRequired[int]

要检索的最近消息数量。

push_notification_config 实例属性

push_notification_config: NotRequired[
    PushNotificationConfig
]

服务器在断开连接时应发送通知的位置。

消息发送参数

基类：TypedDict

message/send方法的参数。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class MessageSendParams(TypedDict):
    """Parameters for message/send method."""

    configuration: NotRequired[MessageSendConfiguration]
    """Send message configuration."""

    message: Message
    """The message being sent to the server."""

    metadata: NotRequired[dict[str, Any]]
    """Extension metadata."""

configuration 实例属性

configuration: NotRequired[MessageSendConfiguration]

发送消息配置。

message 实例属性

message: Message

正在发送到服务器的消息。

metadata 实例属性

metadata: NotRequired[dict[str, Any]]

扩展元数据。

任务发送参数

基类：TypedDict

框架内任务执行的内部参数。

注意：这不是A2A协议的一部分 - 它在内部用于代理/工作单元通信。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class TaskSendParams(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.
    """

    id: str
    """The id of the task."""

    context_id: str
    """The context id for the task."""

    message: Message
    """The message to process."""

    history_length: NotRequired[int]
    """Number of recent messages to be retrieved."""

    metadata: NotRequired[dict[str, Any]]
    """Extension metadata."""

id 实例属性

id: str

任务的ID。

context_id 实例属性

context_id: str

任务的上下文ID。

message 实例属性

message: Message

要处理的消息。

history_length 实例属性

history_length: NotRequired[int]

要检索的最近消息数量。

metadata 实例属性

metadata: NotRequired[dict[str, Any]]

扩展元数据。

列出任务推送通知配置参数

基类：TypedDict

用于获取与任务关联的pushNotificationConfigurations列表的参数。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class ListTaskPushNotificationConfigParams(TypedDict):
    """Parameters for getting list of pushNotificationConfigurations associated with a Task."""

    id: str
    """Task id."""

    metadata: NotRequired[dict[str, Any]]
    """Extension metadata."""

id 实例属性

id: str

任务ID。

metadata 实例属性

metadata: NotRequired[dict[str, Any]]

扩展元数据。

删除任务推送通知配置参数

基类：TypedDict

用于移除与任务关联的pushNotificationConfiguration的参数。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

@pydantic.with_config({'alias_generator': to_camel})
class DeleteTaskPushNotificationConfigParams(TypedDict):
    """Parameters for removing pushNotificationConfiguration associated with a Task."""

    id: str
    """Task id."""

    push_notification_config_id: str
    """The push notification config id to delete."""

    metadata: NotRequired[dict[str, Any]]
    """Extension metadata."""

id 实例属性

id: str

任务ID。

push_notification_config_id 实例属性

push_notification_config_id: str

要删除的推送通知配置ID。

metadata 实例属性

metadata: NotRequired[dict[str, Any]]

扩展元数据。

JSONRPC消息

基类：TypedDict

一个JSON RPC消息。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

class JSONRPCMessage(TypedDict):
    """A JSON RPC message."""

    jsonrpc: Literal['2.0']
    """The JSON RPC version."""

    id: int | str | None
    """The request id."""

jsonrpc 实例属性

jsonrpc: Literal['2.0']

JSON RPC版本。

id 实例属性

id: int | str | None

请求ID。

JSONRPC请求

基类：JSONRPCMessage, Generic[Method, Params]

一个JSON RPC请求。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

class JSONRPCRequest(JSONRPCMessage, Generic[Method, Params]):
    """A JSON RPC request."""

    method: Method
    """The method to call."""

    params: Params
    """The parameters to pass to the method."""

method 实例属性

method: Method

要调用的方法。

params 实例属性

params: Params

传递给方法的参数。

JSONRPC错误

基类：TypedDict, Generic[CodeT, MessageT]

一个JSON RPC错误。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

class JSONRPCError(TypedDict, Generic[CodeT, MessageT]):
    """A JSON RPC error."""

    code: CodeT
    message: MessageT
    data: NotRequired[Any]

JSONRPC响应

基类：JSONRPCMessage, Generic[ResultT, ErrorT]

一个JSON RPC响应。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/schema.py

class JSONRPCResponse(JSONRPCMessage, Generic[ResultT, ErrorT]):
    """A JSON RPC response."""

    result: NotRequired[ResultT]
    error: NotRequired[ErrorT]

JSONParseError 模块属性

JSONParseError = JSONRPCError[
    Literal[-32700], Literal["Invalid JSON payload"]
]

用于解析错误的JSON RPC错误。

InvalidRequestError 模块属性

InvalidRequestError = JSONRPCError[
    Literal[-32600],
    Literal["Request payload validation error"],
]

用于无效请求的JSON RPC错误。

MethodNotFoundError 模块属性

MethodNotFoundError = JSONRPCError[
    Literal[-32601], Literal["Method not found"]
]

用于方法未找到的JSON RPC错误。

InvalidParamsError 模块属性

InvalidParamsError = JSONRPCError[
    Literal[-32602], Literal["Invalid parameters"]
]

用于无效参数的JSON RPC错误。

InternalError 模块属性

InternalError = JSONRPCError[
    Literal[-32603], Literal["Internal error"]
]

用于内部错误的JSON RPC错误。

TaskNotFoundError 模块属性

TaskNotFoundError = JSONRPCError[
    Literal[-32001], Literal["Task not found"]
]

用于任务未找到的JSON RPC错误。

TaskNotCancelableError 模块属性

TaskNotCancelableError = JSONRPCError[
    Literal[-32002], Literal["Task not cancelable"]
]

用于任务不可取消的JSON RPC错误。

PushNotificationNotSupportedError 模块属性

PushNotificationNotSupportedError = JSONRPCError[
    Literal[-32003],
    Literal["Push notification not supported"],
]

用于不支持推送通知的JSON RPC错误。

UnsupportedOperationError 模块属性

UnsupportedOperationError = JSONRPCError[
    Literal[-32004],
    Literal["This operation is not supported"],
]

用于不支持的操作的JSON RPC错误。

ContentTypeNotSupportedError 模块属性

ContentTypeNotSupportedError = JSONRPCError[
    Literal[-32005], Literal["Incompatible content types"]
]

用于不兼容内容类型的JSON RPC错误。

InvalidAgentResponseError 模块属性

InvalidAgentResponseError = JSONRPCError[
    Literal[-32006], Literal["Invalid agent response"]
]

用于无效智能体响应的JSON RPC错误。

SendMessageRequest 模块属性

SendMessageRequest = JSONRPCRequest[
    Literal["message/send"], MessageSendParams
]

用于发送消息的JSON RPC请求。

SendMessageResponse 模块属性

SendMessageResponse = JSONRPCResponse[
    Union[Task, Message], JSONRPCError[Any, Any]
]

发送消息的JSON RPC响应。

StreamMessageRequest 模块属性

StreamMessageRequest = JSONRPCRequest[
    Literal["message/stream"], MessageSendParams
]

用于流式传输消息的JSON RPC请求。

GetTaskRequest 模块属性

GetTaskRequest = JSONRPCRequest[
    Literal["tasks/get"], TaskQueryParams
]

用于获取任务的JSON RPC请求。

GetTaskResponse 模块属性

GetTaskResponse = JSONRPCResponse[Task, TaskNotFoundError]

获取任务的JSON RPC响应。

CancelTaskRequest 模块属性

CancelTaskRequest = JSONRPCRequest[
    Literal["tasks/cancel"], TaskIdParams
]

用于取消任务的JSON RPC请求。

CancelTaskResponse 模块属性

CancelTaskResponse = JSONRPCResponse[
    Task, Union[TaskNotCancelableError, TaskNotFoundError]
]

取消任务的JSON RPC响应。

SetTaskPushNotificationRequest 模块属性

SetTaskPushNotificationRequest = JSONRPCRequest[
    Literal["tasks/pushNotification/set"],
    TaskPushNotificationConfig,
]

用于设置任务推送通知的JSON RPC请求。

SetTaskPushNotificationResponse 模块属性

SetTaskPushNotificationResponse = JSONRPCResponse[
    TaskPushNotificationConfig,
    PushNotificationNotSupportedError,
]

设置任务推送通知的JSON RPC响应。

GetTaskPushNotificationRequest 模块属性

GetTaskPushNotificationRequest = JSONRPCRequest[
    Literal["tasks/pushNotification/get"], TaskIdParams
]

用于获取任务推送通知的JSON RPC请求。

GetTaskPushNotificationResponse 模块属性

GetTaskPushNotificationResponse = JSONRPCResponse[
    TaskPushNotificationConfig,
    PushNotificationNotSupportedError,
]

获取任务推送通知的JSON RPC响应。

ResubscribeTaskRequest 模块属性

ResubscribeTaskRequest = JSONRPCRequest[
    Literal["tasks/resubscribe"], TaskIdParams
]

用于重新订阅任务的JSON RPC请求。

ListTaskPushNotificationConfigRequest 模块属性

ListTaskPushNotificationConfigRequest = JSONRPCRequest[
    Literal["tasks/pushNotificationConfig/list"],
    ListTaskPushNotificationConfigParams,
]

用于列出任务推送通知配置的JSON RPC请求。

DeleteTaskPushNotificationConfigRequest 模块属性

DeleteTaskPushNotificationConfigRequest = JSONRPCRequest[
    Literal["tasks/pushNotificationConfig/delete"],
    DeleteTaskPushNotificationConfigParams,
]

用于删除任务推送通知配置的JSON RPC请求。

A2ARequest 模块属性

A2ARequest = Annotated[
    Union[
        SendMessageRequest,
        StreamMessageRequest,
        GetTaskRequest,
        CancelTaskRequest,
        SetTaskPushNotificationRequest,
        GetTaskPushNotificationRequest,
        ResubscribeTaskRequest,
        ListTaskPushNotificationConfigRequest,
        DeleteTaskPushNotificationConfigRequest,
    ],
    Discriminator("method"),
]

向A2A服务器发出的JSON RPC请求。

A2AResponse 模块属性

A2AResponse: TypeAlias = Union[
    SendMessageResponse,
    GetTaskResponse,
    CancelTaskResponse,
    SetTaskPushNotificationResponse,
    GetTaskPushNotificationResponse,
]

来自A2A服务器的JSON RPC响应。

A2A客户端

A2A协议的客户端。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/client.py

class A2AClient:
    """A client for the A2A protocol."""

    def __init__(self, base_url: str = 'https://:8000', http_client: httpx.AsyncClient | None = None) -> None:
        if http_client is None:
            self.http_client = httpx.AsyncClient(base_url=base_url)
        else:
            self.http_client = http_client
            self.http_client.base_url = base_url

    async def send_message(
        self,
        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.
        """
        params = MessageSendParams(message=message)
        if metadata is not None:
            params['metadata'] = metadata
        if configuration is not None:
            params['configuration'] = configuration

        request_id = str(uuid.uuid4())
        payload = SendMessageRequest(jsonrpc='2.0', id=request_id, method='message/send', params=params)
        content = send_message_request_ta.dump_json(payload, by_alias=True)
        response = await self.http_client.post('/', content=content, headers={'Content-Type': 'application/json'})
        self._raise_for_status(response)

        return send_message_response_ta.validate_json(response.content)

    async def get_task(self, task_id: str) -> GetTaskResponse:
        payload = GetTaskRequest(jsonrpc='2.0', id=None, method='tasks/get', params={'id': task_id})
        content = a2a_request_ta.dump_json(payload, by_alias=True)
        response = await self.http_client.post('/', content=content, headers={'Content-Type': 'application/json'})
        self._raise_for_status(response)
        return get_task_response_ta.validate_json(response.content)

    def _raise_for_status(self, response: httpx.Response) -> None:
        if response.status_code >= 400:
            raise UnexpectedResponseError(response.status_code, response.text)

send_message 异步

send_message(
    message: Message,
    *,
    metadata: dict[str, Any] | None = None,
    configuration: MessageSendConfiguration | None = None
) -> SendMessageResponse

使用A2A协议发送一条消息。

返回一个包含结果（任务）或错误的JSON-RPC响应。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/client.py

async def send_message(
    self,
    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.
    """
    params = MessageSendParams(message=message)
    if metadata is not None:
        params['metadata'] = metadata
    if configuration is not None:
        params['configuration'] = configuration

    request_id = str(uuid.uuid4())
    payload = SendMessageRequest(jsonrpc='2.0', id=request_id, method='message/send', params=params)
    content = send_message_request_ta.dump_json(payload, by_alias=True)
    response = await self.http_client.post('/', content=content, headers={'Content-Type': 'application/json'})
    self._raise_for_status(response)

    return send_message_response_ta.validate_json(response.content)

意外响应错误

基类：Exception

当从服务器收到意外响应时引发的错误。

源代码位于 .venv/lib/python3.12/site-packages/fasta2a/client.py

class UnexpectedResponseError(Exception):
    """An error raised when an unexpected response is received from the server."""

    def __init__(self, status_code: int, content: str) -> None:
        self.status_code = status_code
        self.content = content