Skip to main content

transports.agui

AG-UI protocol adapter.

Implements the AG-UI (Agent UI) protocol for agent-runtimes using Pydantic AI's built-in AG-UI support from pydantic_ai.ui.ag_ui.

Protocol Reference: https://ai.pydantic.dev/ui/ag-ui/

AG-UI is a lightweight protocol focused on UI integration with:

  • Simple JSON message format
  • UI-focused events (text, thinking, tool_use)
  • Real-time streaming support
  • Lightweight for browser clients
  • Identity context support for OAuth token propagation

AGUITransport Objects​

class AGUITransport(BaseTransport)

AG-UI (Agent UI) protocol adapter.

Wraps Pydantic AI's built-in AG-UI support to expose agents through the AG-UI protocol as an ASGI/Starlette application.

The adapter creates a Starlette app that can be mounted into a FastAPI application or run standalone.

This adapter supports per-request model override by extracting the model from the request body and passing it to the AGUIAdapter.

Example:

from pydantic_ai import Agent from agent_runtimes.agents import PydanticAIAgent from agent_runtimes.transports import AGUITransport

Create Pydantic AI agent

pydantic_agent = Agent("openai:gpt-4o")

Wrap with agent adapter

agent = PydanticAIAgent(pydantic_agent)

Create AG-UI adapter

agui_adapter = AGUITransport(agent)

Get the Starlette app

app = agui_adapter.get_app()

Mount in FastAPI

from fastapi import FastAPI from starlette.routing import Mount

main_app = FastAPI() main_app.mount("/agentic_chat", app)

__init__​

def __init__(agent: BaseAgent, agent_id: str | None = None, **kwargs: Any)

Initialize the AG-UI adapter.

Arguments:

  • agent - The agent to adapt.
  • agent_id - Agent ID for usage tracking.
  • **kwargs - Additional arguments passed to AGUIAdapter.dispatch_request.

protocol_name​

@property
def protocol_name() -> str

Get the protocol name.

get_app​

def get_app() -> "Starlette"

Get the Starlette/ASGI application for AG-UI.

This creates a custom Starlette app that supports per-request model override by extracting the model from the request body.

Returns:

Starlette application implementing the AG-UI protocol.

handle_request​

async def handle_request(request: dict[str, Any]) -> dict[str, Any]

Handle an AG-UI request.

Note: AG-UI is primarily a streaming protocol. For direct request handling, use the Starlette app via get_app() instead.

Arguments:

  • request - AG-UI request data.

Returns:

AG-UI response data.

handle_stream​

async def handle_stream(
        request: dict[str, Any]) -> AsyncIterator[dict[str, Any]]

Handle a streaming AG-UI request.

Note: AG-UI uses Starlette/ASGI for streaming. Use get_app() instead.

Arguments:

  • request - AG-UI request data.

    Yields​

    dict[str, Any] AG-UI stream events (not implemented - raises NotImplementedError).