# RemoteWorkflow (/reference/workflows/remote-workflow) `RemoteWorkflow` runs workflows hosted on a remote AgentOS instance or A2A-compatible server. ## 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. Its setup installs the server dependencies, sets `OPENAI_API_KEY`, and registers `qa-workflow`. Use the same `OS_SECURITY_KEY` in the client terminal. ## Basic Usage [#basic-usage] ```python remote_workflow.py 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()) ``` Unless labeled as complete programs, later snippets are fragments for this `async main()` with the same `workflow` 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"`) | | `workflow_id` | `str` | Required | ID of the remote workflow 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. Properties and configuration methods do not forward the run's `auth_token`; they require metadata access without a bearer credential. For the protected example server, use `AgentOSClient.aget_workflow()` with explicit Authorization headers. See [client reference](/reference/clients/agentos-client). ### `id` [#id] Returns the workflow ID. ```python print(workflow.id) # "qa-workflow" ``` ### `name` [#name] Returns the workflow's name from the remote configuration. ```python print(workflow.name) # "QA Workflow" ``` ### `description` [#description] Returns the workflow's description from the remote configuration. ```python print(workflow.description) # "A Q&A workflow for answering questions" ``` ### `db` [#db] Returns a `RemoteDb` instance if the workflow has a database configured. ```python if workflow.db: print(f"Database ID: {workflow.db.id}") ``` ## Methods [#methods] ### `arun` [#arun] Execute the remote workflow asynchronously. ```python # 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) ``` **Parameters:** | Parameter | Type | Default | Description | | ----------------- | ---------------------------------- | -------- | --------------------------------------- | | `input` | `str \| Dict \| List \| BaseModel` | Required | The input for the workflow | | `additional_data` | `Optional[Dict]` | `None` | Additional data to pass to the workflow | | `user_id` | `Optional[str]` | `None` | User ID for the run | | `run_id` | `Optional[str]` | `None` | Custom run ID | | `session_id` | `Optional[str]` | `None` | Session ID for context persistence | | `session_state` | `Optional[Dict]` | `None` | Session state dictionary | | `images` | `Optional[List[Image]]` | `None` | Images to include | | `audio` | `Optional[List[Audio]]` | `None` | Audio to include | | `videos` | `Optional[List[Video]]` | `None` | Videos to include | | `files` | `Optional[List[File]]` | `None` | Files to include | | `stream` | `bool` | `False` | Whether to stream the response | | `stream_events` | `Optional[bool]` | `None` | Whether to stream events | | `auth_token` | `Optional[str]` | `None` | JWT token for authentication | **Returns:** * `WorkflowRunOutput` when `stream=False` * `AsyncIterator[WorkflowRunOutputEvent]` when `stream=True` ### `acontinue_run` [#acontinue_run] Continue a paused workflow run through the AgentOS protocol. The example below assumes `paused` is the actual `WorkflowRunOutput` from a server workflow with confirmation pauses and persistent storage. The basic QA server does not pause automatically. Keep the original requirements, run ID, session ID, and authenticated identity. Different pause kinds need their matching resolution methods; see [workflow HITL](/workflows/hitl/overview). ```python 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"], ) ``` **Parameters:** | Parameter | Type | Default | Description | | ------------------- | ----------------------------- | ------- | --------------------------------------------------------------------- | | `run_response` | `Optional[WorkflowRunOutput]` | `None` | Paused run output containing the run ID, session ID, and requirements | | `run_id` | `Optional[str]` | `None` | Run ID. Required unless `run_response` supplies one | | `session_id` | `Optional[str]` | `None` | Session ID. Overrides the ID from `run_response` | | `step_requirements` | `Optional[List[Any]]` | `None` | Resolved requirements. Overrides requirements from `run_response` | | `stream` | `bool` | `False` | Whether to stream the continued run | | `auth_token` | `Optional[str]` | `None` | JWT token for authentication | **Returns:** * `WorkflowRunOutput` when `stream=False` * `AsyncIterator[WorkflowRunOutputEvent]` when `stream=True` ### `acancel_run` [#acancel_run] Request cancellation through the AgentOS protocol using a real `saved_run_id` from a still-running execution. A successful request records intent; it does not prove execution has stopped. ```python success = await workflow.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 request succeeded. Errors return `False`. The wrapper cannot supply a `session_id`. For scoped authentication that requires the original session, send direct HTTP `POST /workflows/qa-workflow/runs/{run_id}/cancel?session_id={session_id}` with the same bearer identity. Continuation and cancellation are AgentOS-protocol operations, not A2A operations. ### `get_workflow_config` [#get_workflow_config] Get the workflow configuration from the remote server (always fetches fresh). ```python config = await workflow.get_workflow_config() print(f"Workflow name: {config.name}") print(f"Steps: {config.steps}") ``` **Returns:** `WorkflowResponse` ### `refresh_config` [#refresh_config] Force refresh the cached workflow configuration through the AgentOS protocol. ```python config = await workflow.refresh_config() ``` **Returns:** `WorkflowResponse` ## A2A Protocol Support [#a2a-protocol-support] `RemoteWorkflow` supports the A2A REST and JSON-RPC transports below. The A2A execution path forwards message, media, user identity, and the session as `context_id`. AgentOS-specific run IDs, session state, additional data, and stream-event controls are not forwarded. ### Connecting to Agno A2A Servers [#connecting-to-agno-a2a-servers] Install `uv pip install -U "agno[os,a2a]"`. This fragment requires a separate A2A server exposing `my-workflow` at the displayed path; it is not provided by the Python client example server. See [A2A setup](/agent-os/interfaces/a2a/introduction). ```python 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) ``` ### 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] This gateway configuration fragment requires the two deployed servers below. Discovery has the same unauthenticated metadata limitation described above: ```python 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"), ], ) ``` See [AgentOS Gateway](/agent-os/remote-execution/gateway) for more details. ## Streaming Example [#streaming-example] Complete client program for the protected QA example server. Read the workflow's final content once; this also works for function-only workflows that emit no model-content deltas. ```python remote_workflow_stream.py 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()) ``` For streaming continuation, await `workflow.acontinue_run(..., stream=True)` first, then iterate the returned asynchronous iterator. ## Error Handling [#error-handling] ```python 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}") ``` ## Authentication [#authentication] For authenticated AgentOS instances, pass the `auth_token` parameter: ```python response = await workflow.arun( "Process this request", auth_token="your-jwt-token", ) ``` ## Notes [#notes] Remote Workflows via WebSocket are not yet supported. Use HTTP streaming instead.