Skip to main content
RemoteTeam allows you to run teams that are hosted on a remote AgentOS instance. It provides the same interface as a local team, making it easy to integrate remote teams into your applications.

Installation

pip install agno

Basic Usage

from agno.team import RemoteTeam

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

# Run the team (async)
response = await team.arun("Research the latest AI trends")
print(response.content)

Parameters

ParameterTypeDefaultDescription
base_urlstrRequiredBase URL of the remote AgentOS instance (e.g., "http://localhost:7777")
team_idstrRequiredID of the remote team to execute
timeoutfloat300.0Request timeout in seconds
config_ttlfloat300.0Time-to-live for cached configuration in seconds

Properties

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.
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
response = await team.arun(
    "Research AI trends",
    user_id="user-123",
    session_id="session-456",
)
print(response.content)

# Streaming
async for event in team.arun(
    "Analyze this topic",
    stream=True,
    user_id="user-123",
):
    if hasattr(event, "content") and event.content:
        print(event.content, end="", flush=True)
Parameters:
ParameterTypeDefaultDescription
inputstr | List | Dict | Message | BaseModelRequiredThe input message for the team
streamboolFalseWhether to stream the response
user_idOptional[str]NoneUser ID for the run
session_idOptional[str]NoneSession ID for context persistence
session_stateOptional[Dict]NoneSession state dictionary
imagesOptional[Sequence[Image]]NoneImages to include
audioOptional[Sequence[Audio]]NoneAudio to include
videosOptional[Sequence[Video]]NoneVideos to include
filesOptional[Sequence[File]]NoneFiles to include
stream_eventsOptional[bool]NoneWhether to stream events
retriesOptional[int]NoneNumber of retries
knowledge_filtersOptional[Dict]NoneFilters for knowledge search
add_history_to_contextOptional[bool]NoneAdd history to context
dependenciesOptional[Dict]NoneDependencies dictionary
metadataOptional[Dict]NoneMetadata dictionary
auth_tokenOptional[str]NoneJWT token for authentication
Returns:
  • TeamRunOutput when stream=False
  • AsyncIterator[TeamRunOutputEvent] when stream=True

cancel_run

Cancel a running team execution.
success = await team.cancel_run(run_id="run-123")
if success:
    print("Run cancelled")
Parameters:
ParameterTypeDefaultDescription
run_idstrRequiredID of the run to cancel
auth_tokenOptional[str]NoneJWT token for authentication
Returns: bool - True if successfully cancelled

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 = team.refresh_config()
Returns: TeamResponse

Using in AgentOS Gateway

Remote teams can be registered in a local AgentOS to create a gateway:
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

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

team = RemoteTeam(
    base_url="http://localhost:7777",
    team_id="research-team",
)

print("Response: ", end="", flush=True)
async for event in team.arun(
    "Analyze the current state of AI",
    stream=True,
    user_id="user-123",
):
    if isinstance(event, RunContentEvent):
        print(event.content, end="", flush=True)
    elif isinstance(event, RunCompletedEvent):
        print(f"\n\nCompleted: {event.run_id}")

Error Handling

from agno.exceptions import RemoteServerUnavailableError

try:
    response = await team.arun("Hello")
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",
)