> ## Documentation Index
> Fetch the complete documentation index at: https://agno-v2.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# AgentOSClient

> Python client for interacting with AgentOS API endpoints

The `AgentOSClient` provides a convenient interface for interacting with a running AgentOS instance. It supports all AgentOS operations including running agents, teams, and workflows, managing sessions and memories, and searching knowledge bases.

## Basic Usage

```python theme={null}
from agno.client import AgentOSClient

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

# Get configuration
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]}")
```

## Parameters

| Parameter | Type | Default | Description |
| - | - | - | - |
| `base_url` | `str` | Required | Base URL of the AgentOS instance (e.g., `"http://localhost:7777"`) |
| `timeout` | `float` | `60.0` | Request timeout in seconds |

## Methods

### Discovery & Configuration

#### `aget_config`

Get AgentOS configuration and metadata asynchronously.

```python theme={null}
config = await client.aget_config()
```

**Returns:** `ConfigResponse` containing:

* `os_id`: Unique identifier for the OS instance
* `name`: Name of the OS instance
* `agents`: List of registered agents
* `teams`: List of registered teams
* `workflows`: List of registered workflows
* `interfaces`: List of available interfaces

#### `get_config`

Synchronous version of `aget_config`.

```python theme={null}
config = client.get_config()
```

#### `list_agents`

List all agents configured in the AgentOS instance.

```python theme={null}
agents = await client.list_agents()
for agent in agents:
    print(f"{agent.id}: {agent.name}")
```

**Returns:** `List[AgentSummaryResponse]`

#### `aget_agent`

Get detailed configuration for a specific agent.

```python theme={null}
agent = await client.aget_agent(agent_id="my-agent")
print(f"Name: {agent.name}")
print(f"Model: {agent.model}")
print(f"Tools: {agent.tools}")
```

**Parameters:**

* `agent_id` (str): ID of the agent to retrieve

**Returns:** `AgentResponse`

#### `list_teams`

List all teams configured in the AgentOS instance.

```python theme={null}
teams = await client.list_teams()
```

**Returns:** `List[TeamSummaryResponse]`

#### `aget_team`

Get detailed configuration for a specific team.

```python theme={null}
team = await client.aget_team(team_id="my-team")
```

**Parameters:**

* `team_id` (str): ID of the team to retrieve

**Returns:** `TeamResponse`

#### `list_workflows`

List all workflows configured in the AgentOS instance.

```python theme={null}
workflows = await client.list_workflows()
```

**Returns:** `List[WorkflowSummaryResponse]`

#### `aget_workflow`

Get detailed configuration for a specific workflow.

```python theme={null}
workflow = await client.aget_workflow(workflow_id="my-workflow")
```

**Parameters:**

* `workflow_id` (str): ID of the workflow to retrieve

**Returns:** `WorkflowResponse`

***

### Running Agents

#### `run_agent`

Execute an agent run (non-streaming).

```python theme={null}
result = await client.run_agent(
    agent_id="my-agent",
    message="What is 2 + 2?",
    session_id="session-123",
    user_id="user-456",
)
print(f"Response: {result.content}")
print(f"Tokens: {result.metrics.total_tokens}")
```

**Parameters:**

| Parameter | Type | Default | Description |
| - | - | - | - |
| `agent_id` | `str` | Required | ID of the agent to run |
| `message` | `str` | Required | The message/prompt for the agent |
| `session_id` | `Optional[str]` | `None` | Session ID for context persistence |
| `user_id` | `Optional[str]` | `None` | User ID for the run |
| `images` | `Optional[Sequence[Image]]` | `None` | Images to include |
| `audio` | `Optional[Sequence[Audio]]` | `None` | Audio to include |
| `videos` | `Optional[Sequence[Video]]` | `None` | Videos to include |
| `files` | `Optional[Sequence[File]]` | `None` | Files to include |
| `headers` | `Optional[Dict[str, str]]` | `None` | HTTP headers to include in the request |
| `session_state` | `Optional[Dict]` | `None` | Session state dictionary (passed via `**kwargs`) |
| `dependencies` | `Optional[Dict]` | `None` | Dependencies dictionary (passed via `**kwargs`) |
| `metadata` | `Optional[Dict]` | `None` | Metadata dictionary (passed via `**kwargs`) |
| `knowledge_filters` | `Optional[Dict]` | `None` | Filters for knowledge search (passed via `**kwargs`) |
| `output_schema` | `Optional[Dict]` | `None` | JSON schema for structured output (passed via `**kwargs`) |

**Returns:** `RunOutput`

#### `run_agent_stream`

Stream an agent run response.

```python theme={null}
from agno.run.agent import RunContentEvent, RunCompletedEvent

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"\nRun ID: {event.run_id}")
```

**Parameters:** Same as `run_agent`

**Yields:** `RunOutputEvent` (one of `RunStartedEvent`, `RunContentEvent`, `ToolCallStartedEvent`, `ToolCallCompletedEvent`, `RunCompletedEvent`, etc.)

#### `continue_agent_run`

Continue a paused agent run with tool results.

```python theme={null}
from agno.models.response import ToolExecution

result = await client.continue_agent_run(
    agent_id="my-agent",
    run_id="run-123",
    tools=[
        ToolExecution(
            tool_call_id="call-1",
            result="Tool result here",
        )
    ],
)
```

**Parameters:**

| Parameter | Type | Default | Description |
| - | - | - | - |
| `agent_id` | `str` | Required | ID of the agent |
| `run_id` | `str` | Required | ID of the run to continue |
| `tools` | `List[ToolExecution]` | Required | Tool execution results |
| `session_id` | `Optional[str]` | `None` | Session ID |
| `user_id` | `Optional[str]` | `None` | User ID |
| `headers` | `Optional[Dict[str, str]]` | `None` | HTTP headers to include in the request |

**Returns:** `RunOutput`

#### `cancel_agent_run`

Cancel an agent run.

```python theme={null}
await client.cancel_agent_run(agent_id="my-agent", run_id="run-123")
```

***

### Running Teams

#### `run_team`

Execute a team run (non-streaming).

```python theme={null}
result = await client.run_team(
    team_id="research-team",
    message="Research the latest AI trends",
    user_id="user-123",
)
print(f"Response: {result.content}")
```

**Parameters:**

| Parameter | Type | Default | Description |
| - | - | - | - |
| `team_id` | `str` | Required | ID of the team to run |
| `message` | `str` | Required | The message/prompt for the team |
| `session_id` | `Optional[str]` | `None` | Session ID for context persistence |
| `user_id` | `Optional[str]` | `None` | User ID for the run |
| `images` | `Optional[Sequence[Image]]` | `None` | Images to include |
| `audio` | `Optional[Sequence[Audio]]` | `None` | Audio to include |
| `videos` | `Optional[Sequence[Video]]` | `None` | Videos to include |
| `files` | `Optional[Sequence[File]]` | `None` | Files to include |
| `headers` | `Optional[Dict[str, str]]` | `None` | HTTP headers to include in the request |

**Returns:** `TeamRunOutput`
Paused runs return `status="paused"` and `requirements` on the `TeamRunOutput`.

#### `run_team_stream`

Stream a team run response.

```python theme={null}
from agno.run.team import RunContentEvent

async for event in client.run_team_stream(
    team_id="research-team",
    message="Analyze this topic",
):
    if isinstance(event, RunContentEvent):
        print(event.content, end="", flush=True)
```

**Yields:** `TeamRunOutputEvent`

#### `cancel_team_run`

Cancel a team run.

```python theme={null}
await client.cancel_team_run(team_id="my-team", run_id="run-123")
```

***

### Running Workflows

#### `run_workflow`

Execute a workflow run (non-streaming).

```python theme={null}
result = await client.run_workflow(
    workflow_id="qa-workflow",
    message="What are the benefits of Python?",
    user_id="user-123",
)
print(f"Response: {result.content}")
print(f"Status: {result.status}")
```

**Parameters:**

| Parameter | Type | Default | Description |
| - | - | - | - |
| `workflow_id` | `str` | Required | ID of the workflow to run |
| `message` | `str` | Required | The message/prompt for the workflow |
| `session_id` | `Optional[str]` | `None` | Session ID for context persistence |
| `user_id` | `Optional[str]` | `None` | User ID for the run |
| `images` | `Optional[Sequence[Image]]` | `None` | Images to include |
| `audio` | `Optional[Sequence[Audio]]` | `None` | Audio to include |
| `videos` | `Optional[Sequence[Video]]` | `None` | Videos to include |
| `files` | `Optional[Sequence[File]]` | `None` | Files to include |
| `headers` | `Optional[Dict[str, str]]` | `None` | HTTP headers to include in the request |

**Returns:** `WorkflowRunOutput`

#### `run_workflow_stream`

Stream a workflow run response.

```python theme={null}
async for event in client.run_workflow_stream(
    workflow_id="qa-workflow",
    message="Explain machine learning",
):
    if event.event == "RunContent" and hasattr(event, "content"):
        print(event.content, end="", flush=True)
```

**Yields:** `WorkflowRunOutputEvent`

#### `cancel_workflow_run`

Cancel a workflow run.

```python theme={null}
await client.cancel_workflow_run(workflow_id="my-workflow", run_id="run-123")
```

***

### Memory Operations

#### `create_memory`

Create a new user memory.

```python theme={null}
memory = await client.create_memory(
    memory="User prefers dark mode",
    user_id="user-123",
    topics=["preferences", "ui"],
)
print(f"Created: {memory.memory_id}")
```

**Parameters:**

| Parameter | Type | Default | Description |
| - | - | - | - |
| `memory` | `str` | Required | The memory content to store |
| `user_id` | `str` | Required | User ID to associate with the memory |
| `topics` | `Optional[List[str]]` | `None` | Topics to categorize the memory |
| `db_id` | `Optional[str]` | `None` | Database ID to use |
| `table` | `Optional[str]` | `None` | Table name to use |
| `headers` | `Optional[Dict[str, str]]` | `None` | HTTP headers to include in the request |

**Returns:** `UserMemorySchema`

#### `list_memories`

List user memories with filtering and pagination.

```python theme={null}
memories = await client.list_memories(
    user_id="user-123",
    topics=["preferences"],
    limit=10,
)
for mem in memories.data:
    print(f"{mem.memory_id}: {mem.memory}")
```

**Parameters:**

| Parameter | Type | Default | Description |
| - | - | - | - |
| `user_id` | `Optional[str]` | `None` | Filter by user ID |
| `agent_id` | `Optional[str]` | `None` | Filter by agent ID |
| `team_id` | `Optional[str]` | `None` | Filter by team ID |
| `topics` | `Optional[List[str]]` | `None` | Filter by topics |
| `search_content` | `Optional[str]` | `None` | Search within memory content |
| `limit` | `int` | `20` | Number of memories per page |
| `page` | `int` | `1` | Page number |
| `sort_by` | `str` | `"updated_at"` | Field to sort by |
| `sort_order` | `str` | `"desc"` | Sort order (`asc` or `desc`) |
| `db_id` | `Optional[str]` | `None` | Database ID to use |
| `table` | `Optional[str]` | `None` | Table name to use |
| `headers` | `Optional[Dict[str, str]]` | `None` | HTTP headers to include in the request |

**Returns:** `PaginatedResponse[UserMemorySchema]`

#### `get_memory`

Get a specific memory by ID.

```python theme={null}
memory = await client.get_memory(memory_id="mem-123", user_id="user-123")
```

**Returns:** `UserMemorySchema`

#### `update_memory`

Update an existing memory.

```python theme={null}
updated = await client.update_memory(
    memory_id="mem-123",
    memory="User strongly prefers dark mode",
    user_id="user-123",
    topics=["preferences", "ui", "accessibility"],
)
```

**Returns:** `UserMemorySchema`

#### `delete_memory`

Delete a specific memory.

```python theme={null}
await client.delete_memory(memory_id="mem-123", user_id="user-123")
```

***

### Session Operations

#### `create_session`

Create a new session.

```python theme={null}
from agno.db.base import SessionType

session = await client.create_session(
    session_type=SessionType.AGENT,
    agent_id="my-agent",
    user_id="user-123",
    session_name="My Chat Session",
)
print(f"Session ID: {session.session_id}")
```

**Parameters:**

| Parameter | Type | Default | Description |
| - | - | - | - |
| `session_type` | `SessionType` | `SessionType.AGENT` | Type of session (AGENT, TEAM, WORKFLOW) |
| `session_id` | `Optional[str]` | `None` | Optional session ID (auto-generated if not provided) |
| `user_id` | `Optional[str]` | `None` | User ID to associate with the session |
| `session_name` | `Optional[str]` | `None` | Human-readable session name |
| `session_state` | `Optional[Dict[str, Any]]` | `None` | Initial session state |
| `metadata` | `Optional[Dict[str, Any]]` | `None` | Session metadata |
| `agent_id` | `Optional[str]` | `None` | Agent ID (for agent sessions) |
| `team_id` | `Optional[str]` | `None` | Team ID (for team sessions) |
| `workflow_id` | `Optional[str]` | `None` | Workflow ID (for workflow sessions) |
| `db_id` | `Optional[str]` | `None` | Database ID to use |
| `headers` | `Optional[Dict[str, str]]` | `None` | HTTP headers to include in the request |

**Returns:** `AgentSessionDetailSchema`, `TeamSessionDetailSchema`, or `WorkflowSessionDetailSchema`

#### `get_sessions`

List sessions with filtering and pagination.

```python theme={null}
sessions = await client.get_sessions(
    user_id="user-123",
    session_type=SessionType.AGENT,
    limit=20,
)
for session in sessions.data:
    print(f"{session.session_id}: {session.session_name}")
```

**Returns:** `PaginatedResponse[SessionSchema]`

#### `get_session`

Get a specific session by ID.

```python theme={null}
session = await client.get_session(
    session_id="session-123",
    session_type=SessionType.AGENT,
)
```

**Returns:** `AgentSessionDetailSchema`, `TeamSessionDetailSchema`, or `WorkflowSessionDetailSchema`

#### `get_session_runs`

Get all runs for a specific session.

```python theme={null}
runs = await client.get_session_runs(session_id="session-123")
for run in runs:
    print(f"{run.run_id}: {run.content[:50]}...")
```

**Returns:** `List[RunSchema | TeamRunSchema | WorkflowRunSchema]`

#### `rename_session`

Rename a session.

```python theme={null}
session = await client.rename_session(
    session_id="session-123",
    session_name="My Updated Session Name",
)
```

**Returns:** Session detail schema

#### `delete_session`

Delete a specific session.

```python theme={null}
await client.delete_session(session_id="session-123")
```

***

### Knowledge Operations

#### `upload_knowledge_content`

Upload content to the knowledge base.

```python theme={null}
from agno.media import File

content = await client.upload_knowledge_content(
    name="My Document",
    description="Important documentation",
    file=File(content=b"...", filename="doc.pdf", mime_type="application/pdf"),
)
print(f"Content ID: {content.id}")
```

**Parameters:**

| Parameter | Type | Default | Description |
| - | - | - | - |
| `name` | `Optional[str]` | `None` | Content name |
| `description` | `Optional[str]` | `None` | Content description |
| `url` | `Optional[str]` | `None` | URL to fetch content from |
| `metadata` | `Optional[Dict[str, Any]]` | `None` | Metadata dictionary for the content |
| `file` | `Optional[Union[File, UploadFile]]` | `None` | File object to upload; accepts a FastAPI `UploadFile` |
| `text_content` | `Optional[str]` | `None` | Raw text content |
| `reader_id` | `Optional[str]` | `None` | Reader to use for processing |
| `chunker` | `Optional[str]` | `None` | Chunking strategy |
| `chunk_size` | `Optional[int]` | `None` | Chunk size for processing |
| `chunk_overlap` | `Optional[int]` | `None` | Chunk overlap for processing |
| `db_id` | `Optional[str]` | `None` | Database ID to use |
| `knowledge_id` | `Optional[str]` | `None` | Knowledge instance ID for content isolation |
| `headers` | `Optional[Dict[str, str]]` | `None` | HTTP headers to include in the request |

**Returns:** `ContentResponseSchema`

#### `search_knowledge`

Search the knowledge base.

```python theme={null}
results = await client.search_knowledge(
    query="What is Agno?",
    limit=5,
)
for result in results.data:
    print(f"Score: {result.reranking_score}")
    print(f"Content: {result.content[:100]}...")
```

**Parameters:**

| Parameter | Type | Default | Description |
| - | - | - | - |
| `query` | `str` | Required | Search query string |
| `max_results` | `Optional[int]` | `None` | Maximum results to return |
| `filters` | `Optional[Dict]` | `None` | Filters to apply |
| `search_type` | `Optional[str]` | `None` | Search type (vector, keyword, hybrid) |
| `vector_db_ids` | `Optional[List[str]]` | `None` | Vector DB IDs to search |
| `limit` | `int` | `20` | Number of results per page |
| `page` | `int` | `1` | Page number |
| `db_id` | `Optional[str]` | `None` | Database ID to use |
| `knowledge_id` | `Optional[str]` | `None` | Knowledge instance ID for content isolation |
| `headers` | `Optional[Dict[str, str]]` | `None` | HTTP headers to include in the request |

**Returns:** `PaginatedResponse[VectorSearchResult]`

#### `list_knowledge_content`

List all content in the knowledge base.

```python theme={null}
content = await client.list_knowledge_content(limit=20)
for item in content.data:
    print(f"{item.id}: {item.name}")
```

**Returns:** `PaginatedResponse[ContentResponseSchema]`

#### `get_knowledge_config`

Get knowledge base configuration.

```python theme={null}
config = await client.get_knowledge_config()
print(f"Readers: {config.readers}")
print(f"Chunkers: {config.chunkers}")
```

**Returns:** `KnowledgeConfigResponse`

***

### Trace Operations

#### `get_traces`

List execution traces with filtering and pagination.

```python theme={null}
traces = await client.get_traces(
    agent_id="my-agent",
    limit=20,
)
for trace in traces.data:
    print(f"{trace.trace_id}: {trace.status}")
```

**Returns:** `PaginatedResponse[TraceSummary]`

#### `get_trace`

Get detailed trace information.

```python theme={null}
trace = await client.get_trace(trace_id="trace-123")
print(f"Duration: {trace.duration}")
print(f"Spans: {len(trace.tree)}")
```

**Returns:** `TraceDetail` or `TraceNode` (if span\_id provided)

***

### Metrics Operations

#### `get_metrics`

Retrieve AgentOS metrics and analytics data.

```python theme={null}
from datetime import date

metrics = await client.get_metrics(
    starting_date=date(2024, 1, 1),
    ending_date=date(2024, 1, 31),
)
```

**Returns:** `MetricsResponse`

#### `refresh_metrics`

Manually trigger recalculation of system metrics.

```python theme={null}
metrics = await client.refresh_metrics()
```

**Returns:** `List[DayAggregatedMetrics]`

***

## Error Handling

The client raises `RemoteServerUnavailableError` when the remote server is unavailable:

```python theme={null}
from agno.exceptions import RemoteServerUnavailableError

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

For HTTP errors (4xx, 5xx), the client raises `httpx.HTTPStatusError`.

## Authentication

To include authentication headers in requests, pass the `headers` parameter to any method:

```python theme={null}
headers = {"Authorization": "Bearer your-token"}
config = await client.aget_config(headers=headers)
```


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.