To use FastA2A with your own agentic framework, you must implement the Worker class. The Worker is responsible for executing tasks and typically runs in a separate process or machine from the web server.
Key responsibilities of a Worker implementation:
run_task(params: TaskSendParams): Load the task from storage, update its state to working, load the conversation context, execute your agent logic, and finally update the task with the new message and artifacts while setting the state to completed.cancel_task(params: TaskIdParams): Handle task cancellation.build_message_history(history: list[Message]): Transform stored messages into your agent's internal history format.build_artifacts(result: Any): Convert agent results into a list of Artifact objects.
You can use InMemoryStorage and InMemoryBroker for local development.
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]
class InMemoryWorker(Worker[Context]):
async def run_task(self, params: TaskSendParams) -> None:
# Implementation logic here...
pass
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)
