# Remote Workflow (/agent-os/remote-execution/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 [#prerequisites] Start the [Python client example server](/examples/agent-os/python-client/-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: ```bash 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 [#basic-usage] ```python 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()) ``` ## Streaming Responses [#streaming-responses] 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. ```python 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 [#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. ```python 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 [#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: ```python 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: ```python 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 [#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. ```python 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 [#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: ```python 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 [#error-handling] ```python 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 [#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 [#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. ```python 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 [#reference] For complete API documentation, see [RemoteWorkflow Reference](/reference/workflows/remote-workflow).