Remote Team

Execute teams hosted on remote AgentOS instances

RemoteTeam allows you to execute teams that are running on a remote AgentOS instance. The team runs on the remote server; the client submits requests and receives results.

Prerequisites

Start the Python client example server on port 7778. Follow its server dependency and OPENAI_API_KEY setup; it registers research-team and qa-workflow.

In a separate client terminal, install the client dependencies and set the same OS_SECURITY_KEY used by that server:

uv pip install -U "agno[os]"
export OS_SECURITY_KEY="your_os_security_key_here"

The run examples below send this bearer credential explicitly. The OpenAI key belongs in the server process.

Basic Usage

import os
import asyncio
from agno.team import RemoteTeam

async def main():
    # Connect to a remote team
    team = RemoteTeam(
        base_url="http://localhost:7778",  # Running on localhost for this example
        team_id="research-team",
    )

    # Run the team
    response = await team.arun("Research the latest trends in AI", auth_token=os.environ["OS_SECURITY_KEY"])
    print(response.content)

asyncio.run(main())

Streaming Responses

Stream team responses in real-time:

import os
import asyncio

from agno.team import RemoteTeam
from agno.run.team import RunContentEvent

async def main():
    team = RemoteTeam(
        base_url="http://localhost:7778",  # Running on localhost for this example
        team_id="research-team",
    )

    print("Team Response: ", end="", flush=True)
    async for event in team.arun(
        "Analyze the current state of quantum computing",
        auth_token=os.environ["OS_SECURITY_KEY"],
        stream=True,
    ):
        if isinstance(event, RunContentEvent):
            print(event.content, end="", flush=True)

asyncio.run(main())

Configuration Access

The properties below fetch metadata synchronously when the cache is empty or expired (default TTL: 300 seconds). get_team_config() fetches fresh configuration asynchronously.

These wrapper metadata methods do not accept or forward the run's auth_token. The following cache example requires a trusted deployment where metadata requests are permitted without a bearer credential. For the protected example server, use AgentOSClient with explicit headers instead:

import asyncio
import os

from agno.client import AgentOSClient

async def main():
    client = AgentOSClient(base_url="http://localhost:7778")
    config = await client.aget_team(
        "research-team",
        headers={"Authorization": f"Bearer {os.environ['OS_SECURITY_KEY']}"},
    )
    print(config.name)

asyncio.run(main())

Wrapper cache operations, for a server with the metadata access described above:

import asyncio

from agno.team import RemoteTeam

async def main():
    team = RemoteTeam(
        base_url="http://localhost:7778",  # Running on localhost for this example
        team_id="research-team",
    )

    # Access cached properties
    print(f"Name: {team.name}")
    print(f"Description: {team.description}")
    print(f"Role: {team.role()}")

    # Get fresh configuration
    config = await team.get_team_config()
    print(f"Members: {config.members}")

    # Force refresh cache
    await team.refresh_config()

asyncio.run(main())

Using in Gateway

This configuration pattern assumes two separately deployed servers with the named teams. Save it as gateway.py and run python gateway.py after replacing the hostnames.

Gateway discovery uses the wrappers' metadata methods and therefore has the same credential limitation described above. Per-run credentials do not authenticate discovery; plan upstream metadata access separately before using a protected backend.

from agno.team import RemoteTeam
from agno.os import AgentOS

gateway = AgentOS(
    id="api-gateway",
    teams=[
        RemoteTeam(base_url="http://server-1:7777", team_id="research-team"),
        RemoteTeam(base_url="http://server-2:7777", team_id="analysis-team"),
    ],
)

app = gateway.get_app()

if __name__ == "__main__":
    gateway.serve(app="gateway:app", port=7777)

Authentication

Pass the server credential on each run. This can be the static OS_SECURITY_KEY used above, or an appropriately scoped JWT when the server uses JWT authorization. It does not configure later metadata requests:

import os
import asyncio

from agno.team import RemoteTeam

async def main():
    team = RemoteTeam(
        base_url="http://localhost:7778",  # Running on localhost for this example
        team_id="research-team",
    )

    await team.arun(
        "Research this topic",
        auth_token=os.environ["OS_SECURITY_KEY"],
    )

asyncio.run(main())

Error Handling

import os
import asyncio

from agno.team import RemoteTeam
from agno.exceptions import RemoteServerUnavailableError

async def main():
    team = RemoteTeam(
        base_url="http://localhost:7778",  # Running on localhost for this example
        team_id="research-team",
    )

    try:
        await team.arun("Hello", auth_token=os.environ["OS_SECURITY_KEY"])
    except RemoteServerUnavailableError as e:
        print(f"Cannot connect to server: {e.message}")
        # Handle fallback logic

asyncio.run(main())

A2A Protocol Support

RemoteTeam supports A2A servers implementing its HTTP REST or JSON-RPC binding. Set protocol="a2a"; a2a_protocol="rest" is the default, while JSON-RPC servers require a2a_protocol="json-rpc". Server capabilities determine which operations are available.

Connecting to Agno AgentOS via A2A interface

In the server terminal, install uv pip install -U "agno[a2a]". Enable a2a_interface=True on the example server's AgentOS, then restart it. Use the resource-specific A2A URL. This is an alternative to the default AgentOS REST client above.

import os
import asyncio

from agno.team import RemoteTeam

async def main():
    team = RemoteTeam(
        base_url="http://localhost:7778/a2a/teams/research-team",  # Running on localhost for this example
        team_id="research-team",
        protocol="a2a",
    )

    response = await team.arun("Research the rise of AI in the last decade", auth_token=os.environ["OS_SECURITY_KEY"])
    print(response.content)

    # Streaming is also supported
    async for event in team.arun("Analyze the data on the rise of AI", stream=True, auth_token=os.environ["OS_SECURITY_KEY"]):
        if hasattr(event, "content") and event.content:
            print(event.content, end="", flush=True)

asyncio.run(main())

Developer Resources