Remote Workflow

RemoteWorkflow allows you to execute workflows that are running on a remote AgentOS instance. The workflow 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.workflow import RemoteWorkflow

async def main():
    # Connect to a remote workflow
    workflow = RemoteWorkflow(
        base_url="http://localhost:7778",  # Running on localhost for this example
        workflow_id="qa-workflow",
    )

    # Run the workflow
    response = await workflow.arun("What are the benefits of using Python?", auth_token=os.environ["OS_SECURITY_KEY"])
    print(response.content)
    print(f"Status: {response.status}")

asyncio.run(main())

Stream text emitted by agent or team steps, with a final-content fallback for workflows that emit no text deltas. WorkflowAgentCompleted belongs to the optional workflow agent; WorkflowCompleted marks workflow completion.

The following fragments use await or async for; run them inside an async def function, as in Basic Usage.

import os
from agno.workflow import RemoteWorkflow

workflow = RemoteWorkflow(
    base_url="http://localhost:7778",  # Running on localhost for this example
    workflow_id="qa-workflow",
)

print("Workflow Response: ", end="", flush=True)
printed_content = False
async for event in workflow.arun(
    "Write a story about space exploration",
    auth_token=os.environ["OS_SECURITY_KEY"],
    stream=True,
):
    content = getattr(event, "content", None)
    if event.event in ("RunContent", "TeamRunContent") and content:
        print(content, end="", flush=True)
        printed_content = True
    elif (
        event.event in ("WorkflowCompleted", "WorkflowAgentCompleted")
        and content
        and not printed_content
    ):
        print(content, end="", flush=True)
        printed_content = True

Passing Additional Data

Send additional structured data with the run. The workflow must explicitly consume that data to use it; the example QA workflow answers the message.

import os
from agno.workflow import RemoteWorkflow

workflow = RemoteWorkflow(
    base_url="http://localhost:7778",  # Running on localhost for this example
    workflow_id="qa-workflow",
)

response = await workflow.arun(
    "Analyze the data",
    auth_token=os.environ["OS_SECURITY_KEY"],
    additional_data={
        "metrics": {"revenue": 1000000, "growth": 0.15},
        "period": "Q4 2024",
    },
)

Configuration Access

The properties below fetch metadata synchronously when the cache is empty or expired (default TTL: 300 seconds). get_workflow_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_workflow(
        "qa-workflow",
        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:

from agno.workflow import RemoteWorkflow

workflow = RemoteWorkflow(
    base_url="http://localhost:7778",  # Running on localhost for this example
    workflow_id="qa-workflow",
)

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

# Get fresh configuration
config = await workflow.get_workflow_config()

# Force refresh cache
await workflow.refresh_config()

Using in Gateway

This configuration pattern assumes two separately deployed servers with the named workflows. 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.workflow import RemoteWorkflow
from agno.os import AgentOS

gateway = AgentOS(
    id="api-gateway",
    workflows=[
        RemoteWorkflow(base_url="http://server-1:7777", workflow_id="qa-workflow"),
        RemoteWorkflow(base_url="http://server-2:7777", workflow_id="analysis-workflow"),
    ],
)

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
from agno.workflow import RemoteWorkflow

workflow = RemoteWorkflow(
    base_url="http://localhost:7778",
    workflow_id="qa-workflow",
)

response = await workflow.arun(
    "Process this request",
    auth_token=os.environ["OS_SECURITY_KEY"],
)

Error Handling

import os
from agno.workflow import RemoteWorkflow
from agno.exceptions import RemoteServerUnavailableError

workflow = RemoteWorkflow(
    base_url="http://localhost:7778",
    workflow_id="qa-workflow",
)

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

A2A Protocol Support

RemoteWorkflow 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
from agno.workflow import RemoteWorkflow

workflow = RemoteWorkflow(
    base_url="http://localhost:7778/a2a/workflows/qa-workflow",  # Running on localhost for this example
    workflow_id="qa-workflow",
    protocol="a2a",
)

response = await workflow.arun("Write a story about space exploration", auth_token=os.environ["OS_SECURITY_KEY"])
print(response.content)

# Streaming is also supported
printed_content = False
async for event in workflow.arun("Run the workflow", stream=True, auth_token=os.environ["OS_SECURITY_KEY"]):
    content = getattr(event, "content", None)
    if event.event in ("RunContent", "TeamRunContent") and content:
        print(content, end="", flush=True)
        printed_content = True
    elif (
        event.event in ("WorkflowCompleted", "WorkflowAgentCompleted")
        and content
        and not printed_content
    ):
        print(content, end="", flush=True)
        printed_content = True

Reference

For complete API documentation, see RemoteWorkflow Reference.