> ## Documentation Index
> Fetch the complete documentation index at: https://agno-v2-codex-docs-audit-20260719-0149.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Remote Workflow

> Execute workflows hosted on remote AgentOS instances

`RemoteWorkflow` allows you to execute workflows that are running on a remote AgentOS instance. This enables you to leverage complex multi-step workflows without hosting them locally.

## Prerequisites

You need a running AgentOS instance with at least one workflow configured. See [Run Your AgentOS](/agent-os/run-your-os) to set one up.

## Basic Usage

```python theme={null}
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?")
    print(response.content)
    print(f"Status: {response.status}")

asyncio.run(main())
```

## Streaming Responses

Stream workflow responses in real-time:

```python theme={null}
import asyncio

from agno.workflow import RemoteWorkflow

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

    print("Workflow Response: ", end="", flush=True)
    async for event in workflow.arun(
        "Write a story about space exploration",
        stream=True,
    ):
        # Handle content from agent steps or workflow completion
        if event.event == "RunContent" and hasattr(event, "content"):
            print(event.content, end="", flush=True)
        elif event.event == "WorkflowAgentCompleted" and hasattr(event, "content"):
            if event.content:
                print(event.content, end="", flush=True)

asyncio.run(main())
```

## Passing Additional Data

Send additional structured data to the workflow:

```python theme={null}
import asyncio

from agno.workflow import RemoteWorkflow

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

    await workflow.arun(
        "Analyze the data",
        additional_data={
            "metrics": {"revenue": 1000000, "growth": 0.15},
            "period": "Q4 2024",
        },
    )

asyncio.run(main())
```

## Configuration Access

Access the remote workflow's configuration:

```python theme={null}
import asyncio

from agno.workflow import RemoteWorkflow

async def main():
    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
    await workflow.get_workflow_config()

    # Force refresh cache
    await workflow.refresh_config()

asyncio.run(main())
```

## Using in Gateway

Register remote workflows in an AgentOS gateway:

```python theme={null}
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

For authenticated AgentOS instances:

```python theme={null}
import asyncio

from agno.workflow import RemoteWorkflow

async def main():
    workflow = RemoteWorkflow(
        base_url="http://localhost:7777",
        workflow_id="qa-workflow",
    )

    await workflow.arun(
        "Process this request",
        auth_token="your-jwt-token",
    )

asyncio.run(main())
```

## Error Handling

```python theme={null}
import asyncio

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

async def main():
    workflow = RemoteWorkflow(
        base_url="http://localhost:7777",
        workflow_id="qa-workflow",
    )

    try:
        await workflow.arun("Hello")
    except RemoteServerUnavailableError as e:
        print(f"Cannot connect to server: {e.message}")
        # Handle fallback logic

asyncio.run(main())
```

## A2A Protocol Support

`RemoteWorkflow` can also connect to any A2A-compatible server, enabling communication with workflows built using other frameworks.

### Connecting to Agno AgentOS via A2A interface

```python theme={null}
import asyncio

from agno.workflow import RemoteWorkflow

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

    response = await workflow.arun("Write a story about space exploration")
    print(response.content)

    # Streaming is also supported
    async for event in workflow.arun("Run the workflow", stream=True):
        if event.event == "RunContent" and hasattr(event, "content"):
            print(event.content, end="", flush=True)

asyncio.run(main())
```

## Reference

For complete API documentation, see [RemoteWorkflow Reference](/reference/workflows/remote-workflow).