AgentOS Client

Connect to Agno AgentOS instances via REST API

The AgentOSClient provides a Python interface for interacting with running AgentOS instances. It enables you to:

• Run agents, teams, and workflows programmatically with streaming support
• Manage sessions for conversation persistence across runs
• Search and manage knowledge in connected knowledge bases
• Access memories stored for users
• Monitor traces for debugging and observability

Quick Start

import asyncio
from agno.client import AgentOSClient

async def main():
    # Connect to AgentOS
    client = AgentOSClient(base_url="http://localhost:7777")

    # Get configuration and available agents
    config = await client.aget_config()
    print(f"Connected to: {config.name or config.os_id}")
    print(f"Available agents: {[a.id for a in config.agents]}")

    # Run an agent
    if config.agents:
        result = await client.run_agent(
            agent_id=config.agents[0].id,
            message="Hello, how can you help me?",
        )
        print(f"Response: {result.content}")

asyncio.run(main())

Streaming Responses

Stream responses in real-time for a better user experience:

from agno.client import AgentOSClient
from agno.run.agent import RunContentEvent, RunCompletedEvent

client = AgentOSClient(base_url="http://localhost:7777")

async for event in client.run_agent_stream(
    agent_id="my-agent",
    message="Tell me a story",
):
    if isinstance(event, RunContentEvent):
        print(event.content, end="", flush=True)
    elif isinstance(event, RunCompletedEvent):
        print(f"\nCompleted! Run ID: {event.run_id}")

Authentication

When connecting to authenticated AgentOS instances, pass headers with your requests:

headers = {"Authorization": "Bearer your-jwt-token"}

config = await client.aget_config(headers=headers)
result = await client.run_agent(
    agent_id="my-agent",
    message="Hello",
    headers=headers,
)

Error Handling

from agno.client import AgentOSClient
from agno.exceptions import RemoteServerUnavailableError

client = AgentOSClient(base_url="http://localhost:7777")

try:
    config = await client.aget_config()
except RemoteServerUnavailableError as e:
    print(f"Server unavailable: {e.message}")
    print(f"URL: {e.base_url}")

API Reference

For complete method documentation, parameters, and response types, see the AgentOSClient Reference.

Examples

Basic Client

Connect and explore an AgentOS instance

Run Agents

Execute streaming and non-streaming agent runs

Run Teams

Execute team runs with member coordination

Run Workflows

Execute workflow pipelines

Session Management

Create, list, and manage sessions

Knowledge Search

Search and upload knowledge content

Memory Operations

Create, list, and manage user memories