Skip to main content

routes.agui

AG-UI routes for agent-runtimes server.

TerminateRequest Objects​

class TerminateRequest(BaseModel)

Request to terminate a running agent thread.

TerminateResponse Objects​

class TerminateResponse(BaseModel)

Response from terminate request.

register_thread​

def register_thread(thread_id: str) -> asyncio.Event

Register a running thread and return its cancellation event.

Arguments:

  • thread_id - The unique thread identifier.

Returns:

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

unregister_thread​

def unregister_thread(thread_id: str) -> None

Unregister a thread when it completes.

Arguments:

  • thread_id - The thread identifier to remove.

cancel_thread​

def cancel_thread(thread_id: str) -> bool

Cancel a running thread and interrupt the sandbox.

Arguments:

  • thread_id - The thread identifier to cancel.

Returns:

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

cancel_all_threads​

def cancel_all_threads() -> int

Cancel all running threads and interrupt the sandbox.

Returns:

Number of threads cancelled.

register_agui_agent​

def register_agui_agent(agent_id: str, adapter: AGUITransport) -> None

Register an AG-UI adapter.

Arguments:

  • agent_id - Unique identifier for the agent.
  • adapter - The AGUITransport instance.

unregister_agui_agent​

def unregister_agui_agent(agent_id: str) -> None

Unregister an AG-UI adapter.

Arguments:

  • agent_id - The agent identifier.

get_agui_app​

def get_agui_app(agent_id: str) -> Starlette | None

Get an AG-UI Starlette app by ID.

Arguments:

  • agent_id - The agent identifier.

Returns:

The Starlette app if found, None otherwise.

get_agui_adapter​

def get_agui_adapter(agent_id: str) -> "AGUITransport | None"

Get an AG-UI adapter by ID.

Arguments:

  • agent_id - The agent identifier.

Returns:

The AGUITransport adapter if found, None otherwise.

get_all_agui_adapters​

def get_all_agui_adapters() -> dict[str, "AGUITransport"]

Get all registered AG-UI adapters.

Returns:

Dictionary mapping agent IDs to their AGUITransport adapters.

get_agui_mounts​

def get_agui_mounts() -> list[Mount]

Get all AG-UI mounts for the FastAPI app.

Returns:

List of Starlette Mount objects for each AG-UI agent.

list_agents​

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

List available AG-UI agents.

Returns:

Dictionary with list of agent IDs and their endpoints.

agui_info​

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

Get AG-UI service information.

Returns:

Information about the AG-UI service.

terminate_agent​

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

Terminate a running agent thread or all threads.

This endpoint allows clients to stop running agent executions. If thread_id is provided, only that thread is cancelled. If thread_id is None, all running threads are cancelled.

Arguments:

  • request - Terminate request with optional thread_id.

Returns:

Result of the termination request.

list_threads​

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

List all running AG-UI threads.

Returns:

Dictionary with list of running thread IDs.