AgentOSClient

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

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

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

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",
        )
    ],
)

await client.cancel_agent_run(agent_id="my-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}")

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)

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="my-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}: {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"...", filename="doc.pdf", mime_type="application/pdf"),
)
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.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="my-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_ms}ms")
print(f"Spans: {len(trace.spans)}")

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)