RemoteWorkflow

uv pip install -U 'agno[os]'
export OS_SECURITY_KEY=your_os_security_key_here

import asyncio
import os
from agno.workflow import RemoteWorkflow

async def main():
    # Create a remote workflow pointing to a remote AgentOS instance
    workflow = RemoteWorkflow(
        base_url="http://localhost:7778",
        workflow_id="qa-workflow",
    )

    # Run the workflow (async)
    response = await workflow.arun("What are the benefits of Python?", auth_token=os.environ["OS_SECURITY_KEY"])
    print(response.content)

asyncio.run(main())

print(workflow.id)  # "qa-workflow"

print(workflow.name)  # "QA Workflow"

print(workflow.description)  # "A Q&A workflow for answering questions"

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

# Non-streaming
response = await workflow.arun(
    "Explain machine learning",
    auth_token=os.environ["OS_SECURITY_KEY"],
    user_id="user-123",
    session_id="session-456",
)
print(response.content)
print(f"Status: {response.status}")

# Streaming
async for event in workflow.arun(
    "Generate a report",
    auth_token=os.environ["OS_SECURITY_KEY"],
    stream=True,
    user_id="user-123",
):
    if event.event == "WorkflowCompleted":
        print(event.content)
    elif event.event in {"WorkflowError", "WorkflowCancelled", "WorkflowPaused"}:
        print(event.event)

from agno.run.base import RunStatus

if paused.status != RunStatus.paused or not paused.run_id or not paused.session_id:
    raise ValueError("Expected the original persisted paused workflow run")
for requirement in paused.active_step_requirements:
    if not requirement.needs_confirmation:
        raise ValueError("Resolve this pause using its matching HITL handler")
    if input("Approve this step? [y/N] ").lower() == "y":
        requirement.confirm()
    else:
        requirement.reject()
response = await workflow.acontinue_run(
    run_response=paused,
    auth_token=os.environ["OS_SECURITY_KEY"],
)

success = await workflow.acancel_run(run_id=saved_run_id, auth_token=os.environ["OS_SECURITY_KEY"])
if success:
    print("Cancellation requested")

config = await workflow.get_workflow_config()
print(f"Workflow name: {config.name}")
print(f"Steps: {config.steps}")

config = await workflow.refresh_config()

from agno.workflow import RemoteWorkflow

# Connect to an Agno AgentOS with A2A interface
workflow = RemoteWorkflow(
    base_url="http://localhost:7001/a2a/workflows/my-workflow",
    workflow_id="my-workflow",
    protocol="a2a",
)

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

from agno.workflow import RemoteWorkflow
from agno.os import AgentOS

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

import asyncio
import os
from agno.workflow import RemoteWorkflow

async def main():
    workflow = RemoteWorkflow(base_url="http://localhost:7778", workflow_id="qa-workflow")
    async for event in workflow.arun(
        "Explain machine learning",
        stream=True,
        auth_token=os.environ["OS_SECURITY_KEY"],
    ):
        if event.event == "WorkflowCompleted":
            print(event.content)
        elif event.event in {"WorkflowError", "WorkflowCancelled", "WorkflowPaused"}:
            print(event.event)

asyncio.run(main())

from agno.exceptions import RemoteServerUnavailableError

try:
    response = await workflow.arun("Hello", auth_token=os.environ["OS_SECURITY_KEY"])
except RemoteServerUnavailableError as e:
    print(f"Remote server unavailable: {e.message}")

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