RemoteAgent

RemoteAgent provides an asynchronous network interface to an agent hosted by AgentOS or an A2A-compatible server. Use arun() for streaming or non-streaming execution. The AgentOS protocol also supports acontinue_run() and acancel_run(). Synchronous Agent methods such as run() and print_response() are unavailable.

Installation

uv pip install 'agno[os]'

Basic Usage

Start the remote execution quickstart server first. It exposes assistant-agent on port 7778; only the server needs the documented model key.

import asyncio

from agno.agent import RemoteAgent


async def main():
    agent = RemoteAgent(
        base_url="http://localhost:7778",
        agent_id="assistant-agent",
    )
    response = await agent.arun("What is the capital of France?")
    print(response.content)


asyncio.run(main())

Later examples are fragments inside an async function with the corresponding agent configured. Persistence, tools, knowledge, and authentication require those features on the server.

Parameters

Properties

id

Returns the agent ID.

print(agent.id)

name

Returns the agent's name from the remote configuration.

print(agent.name)

description

Returns the agent's description from the remote configuration.

print(agent.description)

role

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

print(agent.role())

tools

The current tools property returns None for the normal AgentOS response shape. Read the materialized tool list from the remote configuration instead:

config = await agent.get_agent_config()
tools = (config.tools or {}).get("tools", [])

for tool in tools:
    print(tool["name"])

db

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

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

knowledge

Returns a RemoteKnowledge instance if the agent has knowledge configured.

if agent.knowledge:
    print("Agent has knowledge enabled")

Methods

arun

Execute the remote agent asynchronously.

# Non-streaming
response = await agent.arun(
    "Tell me about Python",
    user_id="user-123",
    session_id="session-456",
)
print(response.content)

# Streaming
async for event in agent.arun(
    "Tell me a story",
    stream=True,
    user_id="user-123",
):
    if hasattr(event, "content") and event.content:
        print(event.content, end="", flush=True)

Parameters:

Returns:

• RunOutput when stream=False
• AsyncIterator[RunOutputEvent] when stream=True

acontinue_run

Continue a stored AgentOS run. Paused runs require resolved requirements; the server also supports continuation of eligible non-paused stored runs. The server needs a database, and local AgentOS agents require the original session_id even though the Python argument defaults to None.

For this confirmation-only fragment, paused is the actual paused response from a server with confirmation tools. Preserve its requirement objects and IDs:

from agno.run.base import RunStatus

assert paused.status == RunStatus.paused
assert paused.run_id and paused.session_id and paused.requirements
for requirement in paused.requirements:
    if not requirement.needs_confirmation:
        raise ValueError("Resolve this requirement using its input or execution handler")
    print(requirement.tool_execution)
    if input("Confirm this tool call? [y/N] ").lower() == "y":
        requirement.confirm()
    else:
        requirement.reject()

response = await agent.acontinue_run(
    run_id=paused.run_id,
    session_id=paused.session_id,
    requirements=paused.requirements,
)

Use the corresponding requirement methods for user input, feedback, or external execution; do not manufacture a new tool-call ID.

Parameters:

Returns:

• RunOutput when stream=False
• AsyncIterator[RunOutputEvent] when stream=True

acancel_run

Request cancellation of a running AgentOS execution.

success = await agent.acancel_run(run_id=saved_run_id)
if success:
    print("Cancellation request accepted")

Parameters:

Returns: bool: True when the cancellation request is accepted. This does not prove that execution reached a cancelled terminal state; inspect the run's subsequent events or status.

Use saved_run_id from the active run. The method has no session_id argument. Scoped callers (JWT users with isolation enabled and all non-admin service accounts) must instead use the direct HTTP cancellation pattern with the owned session ID.

get_agent_config

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

config = await agent.get_agent_config()
print(f"Agent name: {config.name}")
print(f"Model: {config.model}")

Returns: AgentResponse

refresh_config

Force refresh the cached agent configuration.

config = await agent.refresh_config()

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

A2A Protocol Support

Install uv pip install 'agno[os,a2a]' for protocol="a2a". REST and JSON-RPC are supported; other transports are not.

Connecting to Agno A2A Servers

Start the A2A introduction server for this fragment.

from agno.agent import RemoteAgent

# Connect to an Agno AgentOS with A2A interface
agent = RemoteAgent(
    base_url="http://localhost:7777/a2a/agents/my_agent",
    agent_id="my_agent",
    protocol="a2a",
)

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

Connecting to Google ADK

This requires a separately running Google ADK server on port 8001. Set a2a_protocol="json-rpc":

from agno.agent import RemoteAgent

# Connect to a Google ADK server
agent = RemoteAgent(
    base_url="http://localhost:8001",
    agent_id="facts_agent",
    protocol="a2a",
    a2a_protocol="json-rpc",
)

response = await agent.arun("Tell me an interesting fact")
print(response.content)

# Streaming is also supported
async for event in agent.arun("Tell me a story", stream=True):
    if hasattr(event, "content") and event.content:
        print(event.content, end="", flush=True)

Protocol Options

Using in AgentOS Gateway

This composition fragment registers remote agents in a local gateway. Replace the illustrative server hostnames and IDs with reachable deployments; see the linked guide for a complete server:

from agno.agent import RemoteAgent
from agno.os import AgentOS

agent_os = AgentOS(
    agents=[
        RemoteAgent(base_url="http://server-1:7777", agent_id="agent-1"),
        RemoteAgent(base_url="http://server-2:7777", agent_id="agent-2"),
    ],
)

See AgentOS Gateway for more details.

Error Handling

from agno.exceptions import RemoteServerUnavailableError

try:
    response = await agent.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 agent.arun(
    "Hello",
    auth_token="your-jwt-token",
)

auth_token applies to run, continuation, and cancellation calls. Configuration and property fetches do not accept a token.