Skip to main content

routes.a2a

A2A (Agent-to-Agent) protocol routes.

Implements the A2A protocol using fasta2a for agent-to-agent communication. This module provides FastA2A application mounts for registered agents.

A2AAgentCard Objects​

@dataclass
class A2AAgentCard()

Agent card configuration for A2A discovery.

A2AAgentRegistration Objects​

@dataclass
class A2AAgentRegistration()

Registration info for an A2A agent.

app​

FastA2A instance

TerminateRequest Objects​

class TerminateRequest(BaseModel)

Request to terminate a running A2A task.

TerminateResponse Objects​

class TerminateResponse(BaseModel)

Response from terminate request.

register_task​

def register_task(task_id: str) -> asyncio.Event

Register a running task and return its cancellation event.

Arguments:

  • task_id - The unique task identifier.

Returns:

An asyncio.Event that can be set to signal cancellation.

unregister_task​

def unregister_task(task_id: str) -> None

Unregister a task when it completes.

Arguments:

  • task_id - The task identifier to remove.

cancel_task​

def cancel_task(task_id: str) -> bool

Cancel a running task.

Arguments:

  • task_id - The task identifier to cancel.

Returns:

True if the task was found and cancelled, False otherwise.

cancel_all_tasks​

def cancel_all_tasks() -> int

Cancel all running tasks.

Returns:

Number of tasks cancelled.

set_a2a_app​

def set_a2a_app(app: Any, api_prefix: str = "/api/v1") -> None

Set the FastAPI app reference for dynamic route mounting.

register_a2a_agent​

def register_a2a_agent(agent: BaseAgent,
                       card: A2AAgentCard,
                       broker: Any | None = None,
                       storage: Any | None = None) -> None

Register an agent with the A2A server.

This creates a FastA2A application for the agent that can be mounted in the FastAPI application.

Arguments:

  • agent - The agent to register (must be a PydanticAIAgent)
  • card - Agent card configuration
  • broker - Optional custom broker (defaults to InMemoryBroker)
  • storage - Optional custom storage (defaults to InMemoryStorage)

unregister_a2a_agent​

def unregister_a2a_agent(agent_id: str) -> None

Unregister an agent from the A2A server.

get_a2a_agents​

def get_a2a_agents() -> dict[str, A2AAgentRegistration]

Get all registered A2A agents.

get_a2a_mounts​

def get_a2a_mounts() -> list[Mount]

Get all A2A mounts for the FastAPI application.

start_a2a_task_managers​

async def start_a2a_task_managers() -> None

Start all A2A TaskManagers.

When FastA2A apps are mounted as sub-applications, their lifespan context managers don't run automatically. We need to manually trigger them.

stop_a2a_task_managers​

async def stop_a2a_task_managers() -> None

Stop all A2A TaskManagers.

Clean up the FastA2A app lifespan contexts.

list_a2a_agents​

@router.get("/agents")
async def list_a2a_agents() -> dict[str, Any]

List all available A2A agents.

Returns agent metadata for discovery. Each agent's full A2A endpoints are available at /api/v1/a2a/agents/{agent_id}/.

terminate_task​

@router.post("/terminate", response_model=TerminateResponse)
async def terminate_task(request: TerminateRequest) -> TerminateResponse

Terminate a running A2A task or all tasks.

This endpoint allows clients to stop running agent executions. If task_id is provided, only that task is cancelled. If task_id is None, all running tasks are cancelled.

Arguments:

  • request - Terminate request with optional task_id.

Returns:

Result of the termination request.

list_tasks​

@router.get("/tasks")
async def list_tasks() -> dict[str, Any]

List all running A2A tasks.

Returns:

Dictionary with list of running task IDs.