RemoteTeam

RemoteTeam allows you to run teams that are hosted on a remote AgentOS instance. It exposes an asynchronous remote execution subset: use arun(), not local run() or print_response().

Installation

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

Start the Python client example server on port 7778 first. Follow its server dependency and OPENAI_API_KEY setup; it registers research-team. The client uses that server's OS_SECURITY_KEY above.

Basic Usage

import asyncio
import os
from agno.team import RemoteTeam

async def main():
    # Create a remote team pointing to a remote AgentOS instance
    team = RemoteTeam(
        base_url="http://localhost:7778",
        team_id="research-team",
    )

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

asyncio.run(main())

Unless labeled as complete programs, later snippets are fragments for this async main() with the same team and os imports.

Parameters

Properties

Metadata access can make synchronous requests when the cache is empty or expired. These properties and configuration methods do not forward a per-run auth_token. Use them only where metadata is accessible without a bearer credential; for the protected example server, follow the authenticated configuration example.

id

Returns the team ID.

print(team.id)  # "research-team"

name

Returns the team's name from the remote configuration.

print(team.name)  # "Research Team"

description

Returns the team's description from the remote configuration.

print(team.description)  # "A team of research specialists"

role

Returns the team's role from the remote configuration. Unlike the other accessors, role is a method, so call it.

print(team.role())  # "researcher"

tools

Returns the team's tools as a list of dictionaries.

tools = team.tools
if tools:
    for tool in tools:
        print(tool["name"])

db

Returns a RemoteDb instance if the team has a database configured.

if team.db:
    print(f"Database ID: {team.db.id}")

knowledge

Returns a RemoteKnowledge instance if the team has knowledge configured.

if team.knowledge:
    print("Team has knowledge enabled")

Methods

arun

Execute the remote team asynchronously.

# Non-streaming
from agno.run.team import RunContentEvent
response = await team.arun(
    "Research AI trends",
    auth_token=os.environ["OS_SECURITY_KEY"],
    user_id="user-123",
    session_id="session-456",
)
print(response.content)

# Streaming
async for event in team.arun(
    "Analyze this topic",
    auth_token=os.environ["OS_SECURITY_KEY"],
    stream=True,
    user_id="user-123",
):
    if isinstance(event, RunContentEvent) and isinstance(event.content, str):
        print(event.content, end="", flush=True)

Parameters:

Returns:

• TeamRunOutput when stream=False
• AsyncIterator[TeamRunOutputEvent] when stream=True

acontinue_run

Continue a paused AgentOS team run with its original requirements (for example, tool approval results). Keep the original run_id, session_id, requirement IDs, and user identity; complete each requirement using its matching confirmation, user-input, or external-execution method. The remote server must persist the paused session. See team HITL examples. This method is unavailable with protocol="a2a".

Parameters:

Returns:

• TeamRunOutput when stream=False
• AsyncIterator[TeamRunOutputEvent] when stream=True

acancel_run

Request cancellation of an AgentOS team execution. A true result means the request was accepted, not that the run has already stopped. The method returns False for A2A and on request errors.

Use an actual saved_run_id returned by a run that is still executing:

success = await team.acancel_run(run_id=saved_run_id, auth_token=os.environ["OS_SECURITY_KEY"])
if success:
    print("Cancellation requested")

Parameters:

Returns: bool — whether the cancellation request succeeded.

This wrapper has no session_id argument. Scoped authentication can require it; use direct HTTP POST /teams/research-team/runs/{run_id}/cancel?session_id={session_id} with the same bearer identity and original session in that case.

get_team_config

Get the team configuration from the remote server (always fetches fresh).

config = await team.get_team_config()
print(f"Team name: {config.name}")
print(f"Members: {config.members}")

Returns: TeamResponse

refresh_config

Force refresh the cached team configuration.

config = await team.refresh_config()

Returns: Optional[TeamResponse] (None when using the A2A protocol)

A2A Protocol Support

RemoteTeam supports the A2A REST and JSON-RPC transports below. The A2A path forwards the message, media, metadata, user identity, and session as context_id; AgentOS-specific history, dependencies, retries, and session-state controls are not forwarded. Continuation and cancellation through this wrapper are unavailable.

Connecting to Agno A2A Servers

Install uv pip install -U "agno[os,a2a]". This fragment requires a separate server with an A2A interface exposing my-team at the displayed path; it is not provided by the Python client example server. See A2A interfaces for server setup.

from agno.team import RemoteTeam

# Connect to an Agno AgentOS with A2A interface
team = RemoteTeam(
    base_url="http://localhost:7001/a2a/teams/my-team",
    team_id="my-team",
    protocol="a2a",
)

response = await team.arun("Hello!")
print(response.content)

Protocol Options

Using in AgentOS Gateway

Remote teams can be registered in a local AgentOS to create a gateway. This configuration fragment requires the two deployed hosts below and metadata access as described above:

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

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

See AgentOS Gateway for more details.

Streaming Example

Complete client program for the protected example server:

import asyncio
import os
from agno.team import RemoteTeam
from agno.run.team import RunContentEvent, RunCompletedEvent

async def main():
    team = RemoteTeam(base_url="http://localhost:7778", team_id="research-team")
    async for event in team.arun(
        "Analyze the current state of AI",
        stream=True,
        user_id="user-123",
        auth_token=os.environ["OS_SECURITY_KEY"],
    ):
        if isinstance(event, RunContentEvent) and isinstance(event.content, str):
            print(event.content, end="", flush=True)
        elif isinstance(event, RunCompletedEvent):
            print(f"\nCompleted: {event.run_id}")

asyncio.run(main())

Content events contain deltas; the completion event may repeat the accumulated answer. Print one representation to avoid duplicated output.

Error Handling

from agno.exceptions import RemoteServerUnavailableError

try:
    response = await team.arun("Hello", auth_token=os.environ["OS_SECURITY_KEY"])
except RemoteServerUnavailableError as e:
    print(f"Remote server unavailable: {e.message}")

Authentication

For authenticated AgentOS instances, pass the auth_token parameter:

response = await team.arun(
    "Research this topic",
    auth_token="your-jwt-token",
)

RemoteTeam also inherits component export methods such as as_tool(); see the MCP component guide for exposure through AgentOS.