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

Prerequisites

Install uv pip install -U "agno[os]" in the client environment. The AgentOS examples below connect to the Remote Execution quickstart server, which registers assistant-agent on port 7778. Follow that guide's server dependency and model-key setup first. Provider credentials belong in the server process.

The local quickstart is unauthenticated; protected servers require a bearer header on each request. Session-management operations additionally require a database registered with the server.

Quick Start

import asyncio
from agno.client import AgentOSClient

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

    # 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 or [])]}")

    # 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:

import asyncio

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

async def main():
    client = AgentOSClient(base_url="http://localhost:7778")

    async for event in client.run_agent_stream(
        agent_id="assistant-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}")

asyncio.run(main())

Authentication

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

import asyncio

from agno.client import AgentOSClient

async def main():
    client = AgentOSClient(base_url="http://localhost:7778")
    headers = {"Authorization": "Bearer your-jwt-token"}

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

asyncio.run(main())

Error Handling

import asyncio

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

async def main():
    client = AgentOSClient(base_url="http://localhost:7778")

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

asyncio.run(main())

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