Session Management

import asyncio
import os

from agno.client import AgentOSClient


HEADERS = {"Authorization": f"Bearer {os.environ['OS_SECURITY_KEY']}"}

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

    # Get available agents
    config = await client.aget_config(headers=HEADERS)
    if not config.agents:
        print("No agents available")
        return

    agent_id = config.agents[0].id
    user_id = "example-user"

    print("=" * 60)
    print("Session Management")
    print("=" * 60)

    # Create a session
    print("\n1. Creating a session...")
    session = await client.create_session(
        agent_id=agent_id,
        user_id=user_id,
        session_name="My Test Session",
        headers=HEADERS,
    )
    print(f"   Session ID: {session.session_id}")
    print(f"   Session Name: {session.session_name}")

    # List sessions
    print("\n2. Listing sessions...")
    sessions = await client.get_sessions(user_id=user_id, headers=HEADERS)
    print(f"   Found {len(sessions.data)} sessions")
    for sess in sessions.data[:5]:
        print(f"   - {sess.session_id}: {sess.session_name or 'Unnamed'}")

    # Run some messages in the session
    print("\n3. Running messages in session...")
    await client.run_agent(
        agent_id=agent_id,
        message="Hello!",
        session_id=session.session_id,
        user_id=user_id,
        headers=HEADERS,
    )
    await client.run_agent(
        agent_id=agent_id,
        message="How are you?",
        session_id=session.session_id,
        user_id=user_id,
        headers=HEADERS,
    )

    # Get session runs
    print("\n4. Getting session runs...")
    runs = await client.get_session_runs(session_id=session.session_id, user_id=user_id, headers=HEADERS)
    print(f"   Found {len(runs)} runs in session")
    for run in runs:
        content_preview = (
            (run.content[:50] + "...")
            if run.content and len(str(run.content)) > 50
            else run.content
        )
        print(f"   - {run.run_id}: {content_preview}")

    # Rename session
    print("\n5. Renaming session...")
    renamed = await client.rename_session(
        session_id=session.session_id,
        user_id=user_id,
        session_name="Renamed Test Session",
        headers=HEADERS,
    )
    print(f"   New name: {renamed.session_name}")

    # Delete session
    print(f"\n6. Deleting session {session.session_id}...")
    await client.delete_session(session.session_id, user_id=user_id, headers=HEADERS)
    print("   Session deleted")


if __name__ == "__main__":
    asyncio.run(main())

uv pip install -U "agno[os]" openai