### Set up FastA2A with InMemoryStorage and InMemoryBroker

Source: https://github.com/datalayer/fasta2a/blob/main/README.md

Complete example of wiring FastA2A with InMemoryStorage, InMemoryBroker, and a custom Worker implementation. The worker must implement run_task and cancel_task; the lifespan context manager starts the task manager and worker. Run with `uvicorn main:app --reload`.

```python
import uuid
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Any

from fasta2a import FastA2A, Worker
from fasta2a.broker import InMemoryBroker
from fasta2a.schema import Artifact, Message, TaskIdParams, TaskSendParams, TextPart
from fasta2a.storage import InMemoryStorage

Context = list[Message]
"""The shape of the context you store in the storage."""


class InMemoryWorker(Worker[Context]):
    async def run_task(self, params: TaskSendParams) -> None:
        task = await self.storage.load_task(params['id'])
        assert task is not None

        await self.storage.update_task(task['id'], state='working')

        context = await self.storage.load_context(task['context_id']) or []
        context.extend(task.get('history', []))

        # Call your agent here...
        message = Message(
            role='agent',
            parts=[TextPart(text=f'Your context is {len(context) + 1} messages long.', kind='text')],
            kind='message',
            message_id=str(uuid.uuid4()),
        )

        # Update the new message to the context.
        context.append(message)

        artifacts = self.build_artifacts(123)
        await self.storage.update_context(task['context_id'], context)
        await self.storage.update_task(task['id'], state='completed', new_messages=[message], new_artifacts=artifacts)

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

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

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


storage = InMemoryStorage()
broker = InMemoryBroker()
worker = InMemoryWorker(storage=storage, broker=broker)


@asynccontextmanager
async def lifespan(app: FastA2A) -> AsyncIterator[None]:
    async with app.task_manager:
        async with worker.run():
            yield


app = FastA2A(storage=storage, broker=broker, lifespan=lifespan)
```

--------------------------------

### Install fasta2a[pydantic-ai] extra

Source: https://github.com/datalayer/fasta2a/blob/main/README.md

Install the Pydantic AI integration extra for FastA2A. Use either pip or uv.

```bash
pip install 'fasta2a[pydantic-ai]'  # or `uv add 'fasta2a[pydantic-ai]'`
```

--------------------------------

### Install fasta2a with pip

Source: https://github.com/datalayer/fasta2a/blob/main/README.md

Install the fasta2a package from PyPI. The comment notes that `uv add fasta2a` is an alternative installation method.

```bash
pip install fasta2a  # or `uv add fasta2a`
```

--------------------------------

### FastA2A Schema Usage Examples

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/api-reference/schema.md

Code examples demonstrating how to validate and serialize FastA2A requests, handle validation errors, and build JSON-RPC error responses.

```APIDOC
## [PYTHON] a2a_request_ta.validate_json

### Description
Validates a raw JSON payload against the FastA2A request schema, returning a typed A2ARequest object. This is used to parse incoming JSON-RPC requests.

### Method
validate_json

### Parameters
- **payload** (dict) - Required - The raw JSON-RPC request payload to validate.

### Request Example
```python
payload = {
    'jsonrpc': '2.0',
    'id': 'abc',
    'method': 'message/send',
    'params': {
        'message': {'role': 'user', 'parts': [{'text': 'hi'}], 'messageId': 'm1'},
    },
}
```

### Response
Returns a validated A2ARequest object (e.g., SendMessageRequest). Access fields using Python-key access (snake_case).

### Response Example
```python
parsed = a2a_request_ta.validate_json(payload)
assert parsed['method'] == 'message/send'
assert parsed['params']['message']['message_id'] == 'm1'
```
```

```APIDOC
## [PYTHON] a2a_request_ta.dump_json

### Description
Serializes a validated A2ARequest object back to the wire format (camelCase). This is used to produce the JSON-RPC request body for sending.

### Method
dump_json

### Parameters
- **parsed** (A2ARequest) - Required - The validated request object to serialize.
- **by_alias** (bool) - Required - Set to True to use camelCase aliases.

### Request Example
```python
wire = a2a_request_ta.dump_json(parsed, by_alias=True)
```

### Response
Returns a JSON string in camelCase wire format.

### Response Example
```python
assert b'"messageId"' in wire
```
```

```APIDOC
## [PYTHON] a2a_request_ta.validate_python

### Description
Validates a Python object (e.g., a dict) against the FastA2A request schema. Raises a ValidationError if the object does not match any discriminated member.

### Method
validate_python

### Parameters
- **data** (dict) - Required - The Python object to validate.

### Request Example
```python
try:
    a2a_request_ta.validate_python({'jsonrpc': '2.0', 'id': 1, 'method': 'nope', 'params': {}})
except ValidationError as exc:
    print('no matching discriminated member:', exc.errors()[0]['type'])
```

### Response
Returns a validated A2ARequest object if successful. Raises ValidationError if the method is not recognized.
```

```APIDOC
## [PYTHON] SendMessageResponse

### Description
Constructs a JSON-RPC error response using the SendMessageResponse model. This is used to build error responses for the message/send method.

### Method
SendMessageResponse

### Parameters
- **jsonrpc** (str) - Required - The JSON-RPC version string.
- **id** (str) - Required - The request identifier.
- **error** (TaskNotFoundError) - Required - The error object containing code and message.

### Request Example
```python
response = SendMessageResponse(
    jsonrpc='2.0',
    id='req-1',
    error=TaskNotFoundError(code=-32001, message='Task not found'),
)
```

### Response
Returns a SendMessageResponse object with the error details.

### Response Example
```python
assert response['error']['code'] == -32001
```
```

--------------------------------

### GET /docs

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/endpoints.md

Serves the bundled docs UI. This endpoint is configurable and disabled when docs_url=None.

```APIDOC
## [GET] /docs

### Description
Serves the bundled docs UI. This endpoint is configurable and disabled when docs_url=None.

### Method
GET

### Endpoint
/docs
```

--------------------------------

### GET /docs

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/endpoints.md

Serves the API documentation page (fasta2a/static/docs.html) as text/html, substituting placeholders with the request's root_path. Disabled when docs_url is None.

```APIDOC
## GET /docs

### Description
Serves the API documentation page (fasta2a/static/docs.html) as text/html, substituting __FASTA2A_API_ROOT__ and __FASTA2A_API_ROOT_JSON__ with the request's root_path. Disabled when docs_url is None.

### Method
GET

### Endpoint
/docs

### Response
#### Success Response (200)
- Content-Type: text/html
- Body: HTML content of the documentation page

#### Response Example
(HTML content of fasta2a/static/docs.html)
```

--------------------------------

### Basic usage with InMemoryBroker

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/api-reference/broker.md

Shows how to use InMemoryBroker as a message pipe. The broker is entered with an async context manager, a task is sent via run_task, and operations are received with receive_task_operations. The example prints the task id 'task-1' and asserts the operation is 'run'.

```python
from fasta2a.broker import InMemoryBroker
from fasta2a.schema import TaskSendParams

broker = InMemoryBroker()

async with broker:  # equivalent to `__aenter__`/`__aexit__`
    await broker.run_task(
        {'id': 'task-1', 'context_id': 'ctx-1', 'message': {'role': 'user', 'parts': [{'text': 'hi'}], 'message_id': 'm1'}}
    )
    async for operation in broker.receive_task_operations():
        assert operation['operation'] == 'run'
        print(operation['params']['id'])  # 'task-1'
        break
```

--------------------------------

### Direct TaskManager usage with missing task

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/api-reference/task_manager.md

Shows direct interaction with TaskManager using in-memory components. This example illustrates error handling for non-existent tasks, stubbed push notifications, and task listing.

```python
from fasta2a.broker import InMemoryBroker
from fasta2a.storage import InMemoryStorage
from fasta2a.task_manager import TaskManager

manager = TaskManager(broker=InMemoryBroker(), storage=InMemoryStorage())
async with manager:
    response = await manager.get_task({'jsonrpc': '2.0', 'id': 'req-1', 'method': 'tasks/get', 'params': {'id': 'missing'}})
    assert response['error']['code'] == -32001  # 'Task not found'

    # push notifications and task listing are stubbed out by design:
    notif = await manager.set_task_push_notification(
        {'jsonrpc': '2.0', 'id': 'req-2', 'method': 'tasks/pushNotification/set', 'params': {'url': 'https://example.com/cb'}}
    )
    assert notif['error']['code'] == -32003
    listed = await manager.list_tasks({'jsonrpc': '2.0', 'id': 'req-3', 'method': 'tasks/list', 'params': {}})
    assert listed['error']['code'] == -32004
```

--------------------------------

### POST / (message/send)

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/endpoints.md

Sends a JSON-RPC message/send request to the server. The example shows a request with a user message and expects a response with a task status of 'submitted'.

```APIDOC
## POST / (message/send)

### Description
Sends a JSON-RPC message/send request to the server. The example shows a request with a user message and expects a response with a task status of 'submitted'.

### Method
POST

### Endpoint
/

### Headers
- Content-Type: application/json

### Request Body
- jsonrpc (string) - Required - JSON-RPC version, e.g., "2.0"
- id (string) - Required - Request identifier
- method (string) - Required - Method name, e.g., "message/send"
- params (object) - Required - Parameters object containing the message
  - message (object) - Required - Message object with role, parts, and messageId

### Request Example
{
  "jsonrpc": "2.0",  "id": "req-1",
  "method": "message/send",  "params": {
    "message": {
      "role": "user",      "parts": [{"text": "hello"}],      "messageId": "m-1"
    }
  }
}

### Response
#### Success Response (200)
- Body: JSON-RPC response with result containing task status.

#### Response Example
{
  "result": {
    "task": {
      "status": {
        "state": "submitted"
      }
    }
  }
}
```

--------------------------------

### GET /.well-known/agent-card.json

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/endpoints.md

Fetches the agent card, which describes the agent's capabilities. The example shows that the card includes a 'capabilities' field with 'streaming' and 'pushNotifications' properties.

```APIDOC
## GET /.well-known/agent-card.json

### Description
Fetches the agent card, which describes the agent's capabilities. The example shows that the card includes a 'capabilities' field with 'streaming' and 'pushNotifications' properties.

### Method
GET

### Endpoint
/.well-known/agent-card.json

### Response
#### Success Response (200)
- Body: JSON object containing agent card information, including 'capabilities'.

#### Response Example
{
  "capabilities": {
    "streaming": true,
    "pushNotifications": false
  }
}
```

--------------------------------

### Using a Fetched Agent Card

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/api-reference/client.md

Initializes the client with fetch_card=True to automatically retrieve and use the agent card configuration.

```python
from fasta2a.client import A2AClient

# `fetch_card=True` performs a synchronous GET of /.well-known/agent-card.json
# and derives the base URL from supported_interfaces[0].url.
client = A2AClient(agent='http://localhost:8000', fetch_card=True)
print(client.agent_card['name'])  # the fetched card is available on the client
```

--------------------------------

### Use extension header helpers (fasta2a.extensions)

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/api-reference/extensions.md

Demonstrates parse_extensions_header, select_activated_extensions, missing_required_extensions, and format_extensions_header with inline comments showing expected outputs. Useful for manual extension negotiation and header formatting.

```python
from fasta2a.extensions import (
    format_extensions_header,
    missing_required_extensions,
    parse_extensions_header,
    select_activated_extensions,
)
from fasta2a.schema import AgentExtension

# "a, a, b " -> ['a', 'b']
parse_extensions_header('a, a, b ')
# ['a', 'b']
select_activated_extensions(
    [AgentExtension(uri='a'), AgentExtension(uri='c')],
    ['a', 'b'],
)
# ['c'] — declared required but not activated
missing_required_extensions(
    [AgentExtension(uri='a', required=True), AgentExtension(uri='c')],
    ['a'],
)
# 'a, b'
format_extensions_header(['a', 'b'])
```

--------------------------------

### TaskManager __aenter__ method

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/api-reference/task_manager.md

Initializes the AsyncExitStack and enters the broker context.

```python
async def __aenter__(self):
    self._aexit_stack = AsyncExitStack()
    await self._aexit_stack.__aenter__()
    await self._aexit_stack.enter_async_context(self.broker)

    return self
```

--------------------------------

### Minimal FastA2A app with custom worker and lifespan

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/api-reference/FastA2A.md

Demonstrates initializing a FastA2A application with an in-memory storage and broker, a custom worker implementation, and a lifespan context manager to manage the task manager and worker lifecycle.

```python
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fasta2a import FastA2A, Worker
from fasta2a.broker import InMemoryBroker
from fasta2a.schema import Artifact, Message, TaskIdParams, TaskSendParams
from fasta2a.storage import InMemoryStorage


class MyWorker(Worker[list[Message]]):
    async def run_task(self, params: TaskSendParams) -> None:
        task = await self.storage.load_task(params['id'])
        assert task is not None
        await self.storage.update_task(task['id'], state='working')
        context = await self.storage.load_context(task['context_id']) or []
        context.append(Message(role='agent', parts=[{'text': 'hi'}], message_id='1'))
        await self.storage.update_context(task['context_id'], context)
        await self.storage.update_task(
            task['id'], state='completed', new_messages=context[-1:], new_artifacts=[]
        )

    async def cancel_task(self, params: TaskIdParams) -> None: ...
    def build_message_history(self, history: list[Message]) -> list[object]: return []
    def build_artifacts(self, result: object) -> list[Artifact]: return []


storage = InMemoryStorage()
broker = InMemoryBroker()
worker = MyWorker(storage=storage, broker=broker)


@asynccontextmanager
async def lifespan(app: FastA2A) -> AsyncIterator[None]:
    async with app.task_manager:
        async with worker.run():
            yield


app = FastA2A(storage=storage, broker=broker, lifespan=lifespan)
```

--------------------------------

### Import pydantic_ai with fallback error

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/api-reference/pydantic_ai.md

Shows the required import guard for the optional `pydantic-ai` extra. Install with `pip install 'fasta2a[pydantic-ai]'` (requires Python >= 3.10).

```python
try:
    from pydantic_ai import (...)
except ImportError as _e:
    raise ImportError(
        'Please install the `pydantic-ai` package to use `fasta2a.pydantic_ai`, '
        "e.g. `pip install 'fasta2a[pydantic-ai]'`"
    ) from _e
```

--------------------------------

### Basic A2AClient Usage

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/api-reference/client.md

Demonstrates initializing the client, sending a message, and polling for the task status.

```python
import asyncio
from fasta2a.client import A2AClient
from fasta2a.schema import Message

async def main() -> None:
    client = A2AClient(agent='http://localhost:8000')

    response = await client.send_message(
        Message(role='user', parts=[{'text': 'What is the weather?'}], message_id='m-1')
    )
    task = response['result']['task']
    print(task['id'], task['status']['state'])  # uuid..., 'submitted'

    # Poll or wait, then fetch the final state:
    final = await client.get_task(task['id'])
    print(final['result']['status']['state'])  # e.g. 'completed'

asyncio.run(main())
```

--------------------------------

### FastA2A Constructor

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/api-reference/FastA2A.md

Initializes a FastA2A application instance. All parameters are keyword-only. Requires a Storage and Broker. Registers routes for the agent card, JSON-RPC methods, and optional docs UI.

```APIDOC
## FastA2A Constructor

### Description
Initializes a FastA2A application instance. All parameters are keyword-only. Requires a Storage and Broker. Registers routes for the agent card, JSON-RPC methods, and optional docs UI.

### Method
Constructor (class instantiation)

### Signature
```python
FastA2A(
    *,
    storage: Storage,
    broker: Broker,
    name: str | None = None,
    url: str = 'http://localhost:8000',
    version: str = '1.0.0',
    description: str | None = None,
    provider: AgentProvider | None = None,
    skills: list[Skill] | None = None,
    extensions: list[AgentExtension] | None = None,
    docs_url: str | None = '/docs',
    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,
)
```

### Parameters
All parameters are keyword-only.

| Parameter | Type | Required | Default | Description |
|-----------|------|----------|---------|-------------|
| storage | `Storage` | yes | — | Task/context persistence. Any `Storage[ContextT]` implementation. |
| broker | `Broker` | yes | — | Task scheduler that delivers operations to a `Worker`. |
| name | `str \| None` | no | `None` | Agent name on the card; `'My Agent'` when `None`. |
| url | `str` | no | `'http://localhost:8000'` | Card interface URL. |
| version | `str` | no | `'1.0.0'` | Agent version on the card. |
| description | `str \| None` | no | `None` | Card description; `'An AI agent exposed as an A2A agent.'` when `None`. |
| provider | `AgentProvider \| None` | no | `None` | Optional `provider` field on the card. |
| skills | `list[Skill] \| None` | no | `None` | Skills declared on the card (`[]` when `None`). |
| extensions | `list[AgentExtension] \| None` | no | `None` | Capabilities declared on the card and negotiated per request (`[]` when `None`). |
| docs_url | `str \| None` | no | `'/docs'` | Route serving the bundled docs UI; disabled when `None`. |
| debug | `bool` | no | `False` | Passed through to `Starlette.__init__`. |
| routes | `Sequence[Route] \| None` | no | `None` | Additional Starlette routes. |
| middleware | `Sequence[Middleware] \| None` | no | `None` | Additional Starlette middleware. |
| exception_handlers | `dict[Any, ExceptionHandler] \| None` | no | `None` | Additional Starlette exception handlers. |
| lifespan | `Lifespan[FastA2A] \| None` | no | `None` | ASGI lifespan; defaults to `_default_lifespan`, which only runs `app.task_manager` (`async with app.task_manager: yield`). |

### Construction Behavior
- Creates `self.task_manager = TaskManager(broker=broker, storage=storage)` — the app requires it, so both `broker` and `storage` must be provided.
- Registers three routes on `self.router`:
  - `/.well-known/agent-card.json` — `HEAD`, `GET`, `OPTIONS` (agent card).
  - `/` — `POST` (all JSON-RPC methods).
  - `docs_url` — `GET`, when not `None`.
- Sets `self.default_input_modes = ['application/json']` and `self.default_output_modes = ['application/json']` (hard-coded; a comment notes there is no reason to support other modes).
- `self._agent_card_json_schema: bytes | None = None` — the serialized card, built and cached on first card request.
```

--------------------------------

### run()

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/api-reference/worker.md

Async context manager that connects to the broker and makes the worker available to receive commands. It runs the worker loop and cancels it on exit.

```APIDOC
## run()

### Description
Async context manager that connects to the broker and makes the worker available to receive commands. It runs the worker loop and cancels it on exit.

### Method
run()

### Parameters
None

### Returns
AsyncIterator[None] - An async context manager that yields while the worker is running.

### Usage Example
```python
@asynccontextmanager
async def lifespan(app: FastA2A) -> AsyncIterator[None]:
    async with app.task_manager:
        async with worker.run():
            yield
```
```

--------------------------------

### POST / (message/stream)

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/endpoints.md

Sends a JSON-RPC message/stream request to the server, expecting an SSE stream response. The example shows a request with a required extension that is refused, resulting in an error response.

```APIDOC
## POST / (message/stream)

### Description
Sends a JSON-RPC message/stream request to the server, expecting an SSE stream response. The example shows a request with a required extension that is refused, resulting in an error response.

### Method
POST

### Endpoint
/

### Headers
- Content-Type: application/json
- A2A-Extensions: (optional) - List of extensions the client supports

### Request Body
- jsonrpc (string) - Required - JSON-RPC version, e.g., "2.0"
- id (string) - Required - Request identifier
- method (string) - Required - Method name, e.g., "message/stream"
- params (object) - Required - Parameters object containing the message
  - message (object) - Required - Message object with role, parts, and messageId

### Request Example
{
  "jsonrpc": "2.0",
  "id": "req-2",
  "method": "message/stream",
  "params": {
    "message": {
      "role": "user",
      "parts": [{"text": "hi"}],
      "messageId": "m-2"
    }
  }
}

### Response
#### Success Response (200)
- Body: SSE stream of JSON-RPC responses.

#### Error Response (200 with error in body)
- Body: JSON-RPC error object with code -32600 and message "Request payload validation error", including missing_required_extensions.

#### Response Example
{"jsonrpc":"2.0","id":"req-2","error":{"code":-32600,"message":"Request payload validation error","data":{"missing_required_extensions":["https://example.com/ext/trace/v1"]}}}
```

--------------------------------

### Import guard for httpx in A2AClient

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/api-reference/client.md

Shows the import guard that raises an ImportError with a message instructing to install httpx when it is missing. This is the exact code executed when the module is imported without httpx.

```python
try:
    import httpx
except ImportError as _import_error:
    raise ImportError(
        'httpx is required to use the A2AClient. Please install it with `pip install httpx`.',
    ) from _import_error
```

--------------------------------

### Basic usage of InMemoryStorage

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/api-reference/storage.md

Demonstrates submitting a task, updating its state with new messages and artifacts, and loading the task history.

```python
from fasta2a.storage import InMemoryStorage
from fasta2a.schema import Artifact, Message

storage = InMemoryStorage()

message = Message(role='user', parts=[{'text': 'hello'}], message_id='m1')
task = await storage.submit_task('ctx-1', message)
print(task['id'], task['status']['state'])  # uuid..., 'submitted'
print(message['task_id'])  # the message was stamped in place

updated = await storage.update_task(
    task['id'],
    state='completed',
    new_messages=[Message(role='agent', parts=[{'text': 'hi'}], message_id='m2')],
    new_artifacts=[Artifact(artifact_id='a1', parts=[{'text': 'hi'}])],
)
print(updated['status']['state'])  # 'completed'
print(await storage.load_task(task['id']))  # the same task with history + artifacts
```

--------------------------------

### GET /.well-known/agent-card.json

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/endpoints.md

Returns the agent card as application/json, built once and cached. The card describes the agent's name, description, version, supported interfaces, skills, input/output modes, and capabilities.

```APIDOC
## [GET] /.well-known/agent-card.json

### Description
Returns the agent card as application/json, built once and cached in self._agent_card_json_schema.

### Method
GET, HEAD, OPTIONS

### Endpoint
/.well-known/agent-card.json

### Response
#### Success Response (200)
- **name** (string) - The agent name from the constructor (default 'My Agent').
- **description** (string) - The agent description from the constructor (default text when None).
- **version** (string) - The agent version from the constructor (default '1.0.0').
- **supportedInterfaces** (array) - Always exactly one: AgentInterface(protocol_binding='JSONRPC', url=<constructor url>, protocol_version='1.0').
- **skills** (array) - Constructor skills, else [].
- **defaultInputModes** (array) - Hard-coded ['application/json'].
- **defaultOutputModes** (array) - Hard-coded ['application/json'].
- **capabilities** (object) - {streaming: true, pushNotifications: false}; extensions added here when the constructor got extensions.
- **provider** (object) - Present only when the constructor provider is set.

#### Response Example
{
  "name": "My Agent",
  "description": "An AI agent exposed as an A2A agent.",
  "version": "1.0.0",
  "supportedInterfaces": [
    {"protocolBinding": "JSONRPC", "url": "http://localhost:8000", "protocolVersion": "1.0"}
  ],
  "skills": [],
  "defaultInputModes": ["application/json"],
  "defaultOutputModes": ["application/json"],
  "capabilities": {"streaming": true, "pushNotifications": false}
}
```

--------------------------------

### Using a typed context

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/api-reference/storage.md

Illustrates initializing InMemoryStorage with a specific type and managing context data.

```python
from fasta2a.schema import Message
from fasta2a.storage import InMemoryStorage

storage: InMemoryStorage[list[Message]] = InMemoryStorage()
await storage.update_context('ctx-1', [Message(role='user', parts=[{'text': 'hi'}], message_id='m1')])
context = await storage.load_context('ctx-1')  # list[Message] | None
```

--------------------------------

### Instantiate InMemoryStorage and InMemoryBroker

Source: https://github.com/datalayer/fasta2a/blob/main/README.md

Use the in-memory implementations of Storage and Broker when you don't need persistence or a real queue. This snippet creates instances of InMemoryStorage and InMemoryBroker from the fasta2a package.

```python
from fasta2a import InMemoryStorage, InMemoryBroker

storage = InMemoryStorage()
broker = InMemoryBroker()
```

--------------------------------

### Skill TypedDict definition

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/types.md

Defines the Skill TypedDict used by FastA2A(skills=...) and agent_to_a2a(skills=...). Fields id, name, description, tags, input_modes, and output_modes are required; examples and security_requirements are optional.

```python
class Skill(TypedDict):
    id: str
    name: str
    description: str
    tags: list[str]
    examples: NotRequired[list[str]]
    input_modes: list[str]
    output_modes: list[str]
    security_requirements: NotRequired[list[SecurityRequirement]]
```

--------------------------------

### Configuring AgentExtensions

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/api-reference/FastA2A.md

Shows how to register agent extensions with FastA2A. If a client omits a required extension, the application returns a JSON-RPC error.

```python
from fasta2a import A2A_EXTENSIONS_HEADER, AgentExtension, FastA2A

app = FastA2A(
    storage=storage,
    broker=broker,
    extensions=[
        AgentExtension(uri='https://example.com/ext/citations/v1', description='Cites its sources'),
        AgentExtension(uri='https://example.com/ext/trace/v1', required=True),
    ],
)

# A client that omits the required trace extension gets a JSON-RPC error
# instead of a task. HTTP status is still 200; look at the error payload:
#   {"error": {"code": -32600, "message": "Request payload validation error",
#              "data": {"missing_required_extensions": ["https://example.com/ext/trace/v1"]}}}
```

--------------------------------

### Read activated extensions in a Worker (fasta2a.worker.Worker)

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/api-reference/extensions.md

Shows how a Worker subclass checks whether a specific extension URI is in the activated set using the activated_extensions helper. The example includes a placeholder for adding citation parts to artifacts.

```python
from fasta2a.worker import Worker
from fasta2a import activated_extensions


class MyWorker(Worker):
    async def run_task(self, params):
        if 'https://example.com/ext/citations/v1' in activated_extensions(params):
            ...  # add citation parts to the artifacts
```

--------------------------------

### A2AClient Constructor

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/api-reference/client.md

Initializes an A2AClient instance. Accepts a base URL or an AgentCard, optionally fetches the agent card, and sets up the HTTP client.

```APIDOC
## A2AClient Constructor

### Description
Initializes an A2AClient instance. Accepts a base URL or an AgentCard, optionally fetches the agent card, and sets up the HTTP client.

### Method
Constructor

### Signature
```python
def __init__(
    self,
    agent: str | AgentCard = 'http://localhost:8000',
    http_client: httpx.AsyncClient | None = None,
    fetch_card: bool = False,
    relative_card_path: str | None = None,
) -> None:
```

### Parameters
- **agent** (str | AgentCard) - Optional - Base URL of the server, or an already-fetched AgentCard. Default: 'http://localhost:8000'.
- **http_client** (httpx.AsyncClient | None) - Optional - Reused client; its base_url is overwritten with the derived base URL. When None, a new httpx.AsyncClient(base_url=...) is created. Default: None.
- **fetch_card** (bool) - Optional - When True and agent is a str, fetch the agent card with a synchronous httpx.get to derive the base URL. Default: False.
- **relative_card_path** (str | None) - Optional - Card path for the fetch; defaults to '/.well-known/agent-card.json'. Default: None.

### Behavior and Errors
- When fetch_card=True and agent is a str: performs httpx.get(agent.rstrip('/') + relative_card_path), calls response.raise_for_status() (raises httpx.HTTPStatusError on HTTP error status), then validates the JSON as AgentCard via agent_card_ta.validate_python. This is a blocking call in an otherwise async API.
- When agent is an AgentCard, the first supported_interfaces entry's url becomes the base URL.
- Throws ValueError('AgentCard has no supported interfaces to determine the base URL from.') when the card has no supported_interfaces.
- A card passed in (or fetched) is kept on self.agent_card; a string URL leaves it None.
- The reused http_client.base_url is mutated (overwritten) — note this also affects the caller's client.

### Attributes
- **agent_card** (AgentCard | None) - The card when provided/fetched, else None.
- **http_client** (httpx.AsyncClient) - The HTTP client used for all calls; base_url points at the agent.
```

--------------------------------

### Catch JSON-RPC errors in consumer code

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/errors.md

Shows how to catch JSON-RPC errors in consumer code. Errors arrive in a 200 HTTP response; check the response's `error` key. The example checks for code -32600 and prints the `missing_required_extensions` data when present.

```python
from fasta2a.client import A2AClient
from fasta2a.schema import Message

client = A2AClient(agent='http://localhost:8000')
response = await client.send_message(Message(role='user', parts=[{'text': 'hi'}], message_id='m-1'))
if 'error' in response:
    error = response['error']
    print(error['code'], error['message'])
    if error['code'] == -32600 and 'data' in error:
        print('missing required extensions:', error['data']['missing_required_extensions'])
else:
    task = response['result']['task']
```

--------------------------------

### agent_to_a2a function signature

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/api-reference/pydantic_ai.md

Signature of `agent_to_a2a`, the replacement for Pydantic AI's removed `Agent.to_a2a()`. It wraps any Pydantic AI 2 agent into a `FastA2A` ASGI app. Defaults: `storage` to `InMemoryStorage()`, `broker` to `InMemoryBroker()`, `name` to `agent.name`, and `lifespan` to `partial(worker_lifespan, worker=worker, agent=agent)` so the default app starts the task manager and runs the worker.

```python
def agent_to_a2a(
    agent: AbstractAgent[AgentDepsT, OutputDataT],
    *,
    storage: Storage | None = None,
    broker: Broker | None = None,
    name: str | None = None,
    url: str = 'http://localhost:8000',
    version: str = '1.0.0',
    description: str | None = None,
    provider: AgentProvider | None = None,
    skills: list[Skill] | None = None,
    extensions: list[AgentExtension] | None = None,
    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,
) -> FastA2A:
    """Create a FastA2A server from a pydantic-ai agent."""
```

--------------------------------

### Implementing a custom remote broker

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/api-reference/broker.md

Shows the minimal implementation of a custom broker by subclassing Broker and implementing its five abstract members: run_task, cancel_task, __aenter__, __aexit__, and receive_task_operations. The TaskManager enters the broker with async with, and the worker polls receive_task_operations.

```python
from collections.abc import AsyncIterator
from typing import Any
from fasta2a.broker import Broker, TaskOperation
from fasta2a.schema import TaskIdParams, TaskSendParams


class RemoteBroker(Broker):
    async def run_task(self, params: TaskSendParams) -> None:
        ...  # e.g. POST to a queue service

    async def cancel_task(self, params: TaskIdParams) -> None:
        ...  # e.g. POST a cancellation

    async def __aenter__(self):
        ...  # connect
        return self

    async def __aexit__(self, exc_type: Any, exc_value: Any, traceback: Any):
        ...  # disconnect

    async def receive_task_operations(self) -> AsyncIterator[TaskOperation]:
        ...  # yield operations as they arrive (round-robin across workers)
```

--------------------------------

### run() async context manager

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/api-reference/worker.md

Defines the worker's `run()` method as an async context manager. It connects to the broker and makes the worker available to receive commands. While entered, it runs the `_loop` task and cancels it on exit.

```python
@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()
```

--------------------------------

### FastA2A constructor parameters

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/api-reference/FastA2A.md

Shows the constructor signature for FastA2A, which accepts keyword-only arguments including storage, broker, and optional agent-card and Starlette parameters. All parameters are keyword-only. Located in fasta2a/applications.py, lines 45-68.

```python
def __init__(
    self,
    *,
    storage: Storage,
    broker: Broker,
    # Agent card
    name: str | None = None,
    url: str = 'http://localhost:8000',
    version: str = '1.0.0',
    description: str | None = None,
    provider: AgentProvider | None = None,
    skills: list[Skill] | None = None,
    extensions: list[AgentExtension] | None = None,
    docs_url: str | None = '/docs',
    # 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,
):
```

--------------------------------

### TaskManager __aexit__ method

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/api-reference/task_manager.md

Tears down the task manager and exits the broker context, raising a RuntimeError if not initialized.

```python
async def __aexit__(self, exc_type: Any, exc_value: Any, traceback: Any):
    if self._aexit_stack is None:
        raise RuntimeError('TaskManager was not properly initialized.')
    await self._aexit_stack.__aexit__(exc_type, exc_value, traceback)
    self._aexit_stack = None
```

--------------------------------

### tasks/list

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/endpoints.md

Lists tasks. Always returns error -32004 'This operation is not supported'.

```APIDOC
## tasks/list

### Description
Lists tasks. Always returns error -32004 'This operation is not supported'.

### Method
tasks/list

### Parameters
#### Request Body (params)
- **context** (object) - Optional - Context filter.
- **status** (object) - Optional - Status filter.
- **timestamp** (object) - Optional - Timestamp filter.
- **paging** (object) - Optional - Paging parameters.

### Response
#### Error Response (200)
- **error.code** (integer) - -32004
- **error.message** (string) - "This operation is not supported"

### Error Handling
- Always returns error -32004.
```

--------------------------------

### Handling unstarted TaskManager errors

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/api-reference/FastA2A.md

Illustrates that failing to initialize the TaskManager via the lifespan context manager results in a RuntimeError when handling HTTP requests.

```python
app = FastA2A(storage=InMemoryStorage(), broker=InMemoryBroker())
# Without a lifespan entering app.task_manager, every HTTP request raises:
# RuntimeError('TaskManager was not properly initialized.')
```

--------------------------------

### A2AClient constructor

Source: https://github.com/datalayer/fasta2a/blob/main/_autodocs/api-reference/client.md

Creates an A2AClient instance. Optionally fetches the agent card and uses a custom HTTP client.

```APIDOC
## A2AClient

### Description
Creates an A2AClient instance. Optionally fetches the agent card and uses a custom HTTP client.

### Parameters
- **agent** (string) - Required - The base URL of the agent.
- **http_client** (httpx.AsyncClient) - Optional - Custom HTTP client.
- **fetch_card** (bool) - Optional - If True, performs a synchronous GET of /.well-known/agent-card.json and derives the base URL from supported_interfaces[0].url.

### Attributes
- **agent_card** (object) - The fetched agent card, if fetch_card=True.

### Example
```python
client = A2AClient(agent='http://localhost:8000', http_client=httpx.AsyncClient(timeout=30.0))
client = A2AClient(agent='http://localhost:8000', fetch_card=True)
print(client.agent_card['name'])
```
```