# RemoteTeam (/reference/teams/remote-team) `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 [#installation] ```bash uv pip install -U 'agno[os]' export OS_SECURITY_KEY=your_os_security_key_here ``` Start the [Python client example server](/examples/agent-os/python-client/-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 [#basic-usage] ```python remote_team.py 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 [#parameters] | Parameter | Type | Default | Description | | -------------- | ----------------------------- | ----------- | --------------------------------------------------------------------------------- | | `base_url` | `str` | Required | Base URL of the remote AgentOS instance (e.g., `"http://localhost:7778"`) | | `team_id` | `str` | Required | ID of the remote team to execute | | `timeout` | `float` | `300.0` | Request timeout in seconds | | `protocol` | `Literal["agentos", "a2a"]` | `"agentos"` | Communication protocol: AgentOS REST API or A2A for cross-framework communication | | `a2a_protocol` | `Literal["json-rpc", "rest"]` | `"rest"` | Transport used when `protocol="a2a"`: JSON-RPC or REST | | `config_ttl` | `float` | `300.0` | Time-to-live for cached configuration in seconds | ## Properties [#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](/agent-os/remote-execution/remote-team#configuration-access). ### `id` [#id] Returns the team ID. ```python print(team.id) # "research-team" ``` ### `name` [#name] Returns the team's name from the remote configuration. ```python print(team.name) # "Research Team" ``` ### `description` [#description] Returns the team's description from the remote configuration. ```python print(team.description) # "A team of research specialists" ``` ### `role` [#role] Returns the team's role from the remote configuration. Unlike the other accessors, `role` is a method, so call it. ```python print(team.role()) # "researcher" ``` ### `tools` [#tools] Returns the team's tools as a list of dictionaries. ```python tools = team.tools if tools: for tool in tools: print(tool["name"]) ``` ### `db` [#db] Returns a `RemoteDb` instance if the team has a database configured. ```python if team.db: print(f"Database ID: {team.db.id}") ``` ### `knowledge` [#knowledge] Returns a `RemoteKnowledge` instance if the team has knowledge configured. ```python if team.knowledge: print("Team has knowledge enabled") ``` ## Methods [#methods] ### `arun` [#arun] Execute the remote team asynchronously. ```python # 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:** | Parameter | Type | Default | Description | | ------------------------------ | --------------------------------------------- | -------- | ---------------------------------- | | `input` | `str \| List \| Dict \| Message \| BaseModel` | Required | The input message for the team | | `stream` | `bool` | `False` | Whether to stream the response | | `user_id` | `Optional[str]` | `None` | User ID for the run | | `session_id` | `Optional[str]` | `None` | Session ID for context persistence | | `session_state` | `Optional[Dict]` | `None` | Session state dictionary | | `images` | `Optional[Sequence[Image]]` | `None` | Images to include | | `audio` | `Optional[Sequence[Audio]]` | `None` | Audio to include | | `videos` | `Optional[Sequence[Video]]` | `None` | Videos to include | | `files` | `Optional[Sequence[File]]` | `None` | Files to include | | `stream_events` | `Optional[bool]` | `None` | Whether to stream events | | `retries` | `Optional[int]` | `None` | Number of retries | | `knowledge_filters` | `Optional[Dict]` | `None` | Filters for knowledge search | | `add_history_to_context` | `Optional[bool]` | `None` | Add history to context | | `add_dependencies_to_context` | `Optional[bool]` | `None` | Add dependencies to context | | `add_session_state_to_context` | `Optional[bool]` | `None` | Add session state to context | | `dependencies` | `Optional[Dict]` | `None` | Dependencies dictionary | | `metadata` | `Optional[Dict]` | `None` | Metadata dictionary | | `auth_token` | `Optional[str]` | `None` | JWT token for authentication | **Returns:** * `TeamRunOutput` when `stream=False` * `AsyncIterator[TeamRunOutputEvent]` when `stream=True` ### `acontinue_run` [#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](/examples/teams/human-in-the-loop/overview). This method is unavailable with `protocol="a2a"`. **Parameters:** | Parameter | Type | Default | Description | | -------------- | --------------- | -------- | ---------------------------------------------------- | | `run_id` | `str` | Required | ID of the run to continue | | `requirements` | `List[Any]` | Required | `RunRequirement` objects with tool execution results | | `stream` | `bool` | `False` | Whether to stream the response | | `user_id` | `Optional[str]` | `None` | User ID | | `session_id` | `Optional[str]` | `None` | Session ID | | `auth_token` | `Optional[str]` | `None` | JWT token for authentication | **Returns:** * `TeamRunOutput` when `stream=False` * `AsyncIterator[TeamRunOutputEvent]` when `stream=True` ### `acancel_run` [#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: ```python success = await team.acancel_run(run_id=saved_run_id, auth_token=os.environ["OS_SECURITY_KEY"]) if success: print("Cancellation requested") ``` **Parameters:** | Parameter | Type | Default | Description | | ------------ | --------------- | -------- | ---------------------------- | | `run_id` | `str` | Required | ID of the run to cancel | | `auth_token` | `Optional[str]` | `None` | JWT token for authentication | **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_team_config] Get the team configuration from the remote server (always fetches fresh). ```python config = await team.get_team_config() print(f"Team name: {config.name}") print(f"Members: {config.members}") ``` **Returns:** `TeamResponse` ### `refresh_config` [#refresh_config] Force refresh the cached team configuration. ```python config = await team.refresh_config() ``` **Returns:** `Optional[TeamResponse]` (`None` when using the A2A protocol) ## A2A Protocol Support [#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 [#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](/agent-os/interfaces/a2a/introduction) for server setup. ```python 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 [#protocol-options] | Protocol | a2a\_protocol | Use Case | | ----------- | ------------- | -------------------------------------------------- | | `"agentos"` | N/A | Default. Connect to Agno AgentOS REST API | | `"a2a"` | `"rest"` | Connect to A2A servers using REST endpoints | | `"a2a"` | `"json-rpc"` | Connect to Google ADK or pure JSON-RPC A2A servers | ## Using in AgentOS Gateway [#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: ```python 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](/agent-os/remote-execution/gateway) for more details. ## Streaming Example [#streaming-example] Complete client program for the protected example server: ```python remote_team_stream.py 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 [#error-handling] ```python 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 [#authentication] For authenticated AgentOS instances, pass the `auth_token` parameter: ```python 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](/agent-os/mcp/mcp) for exposure through AgentOS.