AgentOSClient

import asyncio

from agno.client import AgentOSClient


async def main():
    client = AgentOSClient(base_url="http://localhost:7778")
    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]}")


asyncio.run(main())

config = await client.aget_config()

config = client.get_config()

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

agent = await client.aget_agent(agent_id="assistant-agent")
print(f"Name: {agent.name}")
print(f"Model: {agent.model}")
print(f"Tools: {agent.tools}")

teams = await client.list_teams()

team = await client.aget_team(team_id="my-team")

workflows = await client.list_workflows()

workflow = await client.aget_workflow(workflow_id="my-workflow")

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

from agno.run.agent import RunContentEvent, RunCompletedEvent

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

from agno.run.base import RunStatus

assert paused.status == RunStatus.paused
assert paused.run_id and paused.session_id and paused.requirements
for requirement in paused.requirements:
    if not requirement.needs_confirmation:
        raise ValueError("Resolve this requirement using its input or execution handler")
    print(requirement.tool_execution)
    if input("Confirm this tool call? [y/N] ").lower() == "y":
        requirement.confirm()
    else:
        requirement.reject()

result = await client.continue_agent_run(
    agent_id=paused.agent_id,
    run_id=paused.run_id,
    session_id=paused.session_id,
    tools=[req.tool_execution for req in paused.requirements if req.tool_execution],
)

await client.cancel_agent_run(agent_id="assistant-agent", run_id="run-123")

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

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)

await client.cancel_team_run(team_id="my-team", run_id="run-123")

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}")

from agno.run.workflow import (
    WorkflowCompletedEvent,
    WorkflowErrorEvent,
    WorkflowCancelledEvent,
    WorkflowPausedEvent,
)

async for event in client.run_workflow_stream(
    workflow_id="qa-workflow",
    message="Explain machine learning",
):
    if isinstance(event, WorkflowCompletedEvent):
        print(event.content)
    elif isinstance(event, WorkflowErrorEvent):
        print(f"Workflow failed: {event.error}")
    elif isinstance(event, WorkflowCancelledEvent):
        print(f"Workflow cancelled: {event.reason}")
    elif isinstance(event, WorkflowPausedEvent):
        print(f"Workflow paused: {event.run_id}")

await client.cancel_workflow_run(workflow_id="my-workflow", run_id="run-123")

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

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}")

memory = await client.get_memory(memory_id="mem-123", user_id="user-123")

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

await client.delete_memory(memory_id="mem-123", user_id="user-123")

from agno.db.base import SessionType

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

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}")

session = await client.get_session(
    session_id="session-123",
    session_type=SessionType.AGENT,
)

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

session = await client.rename_session(
    session_id="session-123",
    session_name="My Updated Session Name",
)

await client.delete_session(session_id="session-123")

from agno.media import File

content = await client.upload_knowledge_content(
    name="My Document",
    description="Important documentation",
    file=File(content=b"Agno builds agents, teams, and workflows.", filename="doc.txt", mime_type="text/plain"),
)
print(f"Content ID: {content.id}")

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]}...")

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

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

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

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

from datetime import date

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

metrics = await client.refresh_metrics()

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}")

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

import os

import httpx

async with httpx.AsyncClient(base_url="http://localhost:7778") as http:
    response = await http.post(
        f"/agents/assistant-agent/runs/{saved_run_id}/cancel",
        params={"session_id": saved_session_id},
        headers={"Authorization": f"Bearer {os.environ['AGENTOS_TOKEN']}"},
    )
    response.raise_for_status()
    print("Cancellation request accepted")