Call and stream an Agent on another AgentOS

Call an agent hosted on another AgentOS with RemoteAgent, in both single-shot and streaming mode.

RemoteAgent presents a remote AgentOS Agent through the same asynchronous run shape used by local Agno components.

remote_agent.py

"""
Call and stream an Agent on another AgentOS
===========================================

RemoteAgent presents a remote AgentOS Agent through the same asynchronous run
shape used by local Agno components.

Prerequisites: start `servers/agentos_server.py` and set OPENAI_API_KEY
Run: .venvs/demo/bin/python cookbook/05_agent_os/20_remote/01_remote_agent.py
Try: change either arithmetic prompt and observe the remote calculator call
"""

import asyncio

from agno.agent import RemoteAgent
from agno.run.agent import (
    RunCompletedEvent,
    RunContentEvent,
    RunErrorEvent,
    RunStartedEvent,
)

BASE_URL = "http://127.0.0.1:7780"

# ---------------------------------------------------------------------------
# Create Remote Agent
# ---------------------------------------------------------------------------

remote_assistant = RemoteAgent(
    base_url=BASE_URL,
    agent_id="assistant-agent",
)

# ---------------------------------------------------------------------------
# Run Remote Agent
# ---------------------------------------------------------------------------


async def run_remote_agent() -> None:
    """Run one complete response and one typed remote stream."""
    response = await remote_assistant.arun(
        "Use the calculator to multiply 17 by 23.",
        session_id="remote-agent-session",
    )
    print(f"Non-streaming run: {response.run_id}")
    print(response.content)

    print("\nStreaming response:")
    async for event in remote_assistant.arun(
        "Use the calculator to add 41 and 1, then explain the result.",
        session_id="remote-agent-session",
        stream=True,
        stream_events=True,
    ):
        if isinstance(event, RunStartedEvent):
            print(f"Run started: {event.run_id}")
        elif isinstance(event, RunContentEvent) and event.content is not None:
            print(event.content, end="", flush=True)
        elif isinstance(event, RunCompletedEvent):
            print(f"\nRun completed: {event.run_id}")
        elif isinstance(event, RunErrorEvent):
            raise RuntimeError(event.content or "Remote Agent run failed")


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

Run the Example

1. Set up your virtual environment

   Mac

   uv venv --python 3.12
   source .venv/bin/activate
   
2. Install dependencies

   uv pip install -U "agno[os]" openai
   
3. Export your API keys

   Mac/Linux

   export OPENAI_API_KEY="your_openai_api_key_here"
   
4. Clone Agno

   Clone the pinned Agno source and run the remaining commands from its root:

   git clone https://github.com/agno-agi/agno.git
   cd agno
   git checkout v3.0.4
   
5. Start the remote AgentOS

   Open another terminal in the parent directory where .venv and the cloned agno directory are siblings, then start the upstream on port 7780:

   Mac/Linux

   source .venv/bin/activate
   export OPENAI_API_KEY="your_openai_api_key_here"
   cd agno
   python cookbook/05_agent_os/20_remote/servers/agentos_server.py
   
6. Run the example

   Run the example from the repository root:

   python cookbook/05_agent_os/20_remote/01_remote_agent.py
   

Full source: cookbook/05_agent_os/20_remote/01_remote_agent.py