Skip to main content

transports.clients.ag_ui_client

AG-UI (Agent UI) protocol client.

Provides:

  • HTTP-based communication with AG-UI compatible agent servers
  • Server-Sent Events (SSE) streaming
  • Message history management
  • Tool call handling

AG-UI Protocol Reference: https://docs.ag-ui.com/

Example:

async with AGUIClient("http://localhost:8000/api/v1/ag-ui/agent-1/") as client: async for event in client.run("Hello, agent!"): if event.type == "TEXT_MESSAGE_CONTENT": print(event.delta, end="", flush=True)

AGUIClientError Objects​

class AGUIClientError(Exception)

Error raised during AG-UI client operations.

AGUIEvent Objects​

@dataclass
class AGUIEvent()

Parsed AG-UI event.

delta​

@property
def delta() -> str | None

Get text delta for TEXT_MESSAGE_CONTENT events.

message_id​

@property
def message_id() -> str | None

Get message ID from event.

tool_call_id​

@property
def tool_call_id() -> str | None

Get tool call ID from event.

tool_name​

@property
def tool_name() -> str | None

Get tool name from TOOL_CALL_START events.

tool_args​

@property
def tool_args() -> str | None

Get tool arguments delta from TOOL_CALL_ARGS events.

tool_result​

@property
def tool_result() -> str | None

Get tool result from TOOL_CALL_RESULT events.

error​

@property
def error() -> str | None

Get error message from RUN_ERROR events.

AGUIConversation Objects​

@dataclass
class AGUIConversation()

Manages conversation state for AG-UI client.

add_user_message​

def add_user_message(content: str) -> UserMessage

Add a user message to the conversation.

add_assistant_message​

def add_assistant_message(content: str,
                          message_id: str | None = None) -> AssistantMessage

Add an assistant message to the conversation.

add_tool_message​

def add_tool_message(tool_call_id: str, content: str) -> ToolMessage

Add a tool result message to the conversation.

clear​

def clear() -> None

Clear conversation history.

AGUIClient Objects​

class AGUIClient()

AG-UI (Agent UI) protocol client.

Connects to AG-UI compatible agent servers via HTTP with SSE streaming.

Example:

async with AGUIClient("http://localhost:8000/api/v1/ag-ui/agent-1/") as client: async for event in client.run("Hello, agent!"): if event.type == EventType.TEXT_MESSAGE_CONTENT: print(event.delta, end="", flush=True)

__init__​

def __init__(url: str,
             headers: dict[str, str] | None = None,
             timeout: float = 60.0,
             model: str | None = None)

Initialize the AG-UI client.

Arguments:

  • url - HTTP URL of the AG-UI endpoint (e.g., http://localhost:8000/api/v1/ag-ui/agent-1/).
  • headers - Optional headers for HTTP requests.
  • timeout - Request timeout in seconds.
  • model - Optional model to use (overrides server default).

connect​

async def connect() -> "AGUIClient"

Initialize the HTTP client.

Returns:

Self for chaining.

disconnect​

async def disconnect() -> None

Close the HTTP client.

__aenter__​

async def __aenter__() -> "AGUIClient"

Async context manager entry.

__aexit__​

async def __aexit__(exc_type: Any, exc_val: Any, exc_tb: Any) -> None

Async context manager exit.

run​

async def run(
    input_text: str,
    context: list[dict[str, Any]] | None = None,
    tools: list[dict[str, Any]] | None = None,
    identities: list[dict[str, Any]] | None = None
) -> AsyncGenerator[AGUIEvent, None]

Run the agent on the given input with SSE streaming.

Arguments:

  • input_text - The input text to send to the agent.
  • context - Optional context items (files, snippets, etc.).
  • tools - Optional tool definitions for the agent.
  • identities - Optional OAuth identities for tool authentication.

Yields:

AGUIEvent objects for each SSE event.

send_message​

async def send_message(content: str,
                       **kwargs: Any) -> AsyncGenerator[AGUIEvent, None]

Send a message to the agent and stream responses.

Convenience method that wraps run().

Arguments:

  • content - The message content.
  • **kwargs - Additional arguments passed to run().

Yields:

AGUIEvent objects.

run_with_tools​

async def run_with_tools(input_text: str, tool_handler: Any,
                         **kwargs: Any) -> AsyncGenerator[AGUIEvent, None]

Run the agent with automatic tool execution.

Arguments:

  • input_text - The input text to send to the agent.
  • tool_handler - Async function to handle tool calls: (tool_name, tool_call_id, args) -> result
  • **kwargs - Additional arguments passed to run().

Yields:

AGUIEvent objects including tool execution results.

new_conversation​

def new_conversation() -> None

Start a new conversation (clears history).

conversation​

@property
def conversation() -> AGUIConversation

Get the current conversation.

messages​

@property
def messages() -> list[Message]

Get conversation messages.

thread_id​

@property
def thread_id() -> str

Get the current thread ID.

connect_agui​

async def connect_agui(url: str,
                       headers: dict[str, str] | None = None,
                       timeout: float = 60.0,
                       model: str | None = None) -> AGUIClient

Connect to an AG-UI compatible agent server.

Convenience function for creating and connecting an AG-UI client.

Arguments:

  • url - HTTP URL of the AG-UI endpoint.
  • headers - Optional headers for HTTP requests.
  • timeout - Request timeout in seconds.
  • model - Optional model to use.

Returns:

Connected AG-UI client.

Example:

client = await connect_agui("http://localhost:8000/api/v1/ag-ui/my-agent/") async for event in client.run("Hello!"): print(event) await client.disconnect()