Execute Workflow

Example workflow server

Install the runtime and SQLite adapter, then configure a shared key:

uv pip install -U "agno[os]" sqlalchemy
export OS_SECURITY_KEY="your-agentos-key"

Save as workflow_api.py and run python workflow_api.py. This workflow runs a Python function and needs no model credentials.

from agno.db.sqlite import SqliteDb
from agno.os import AgentOS
from agno.workflow import Step, StepInput, StepOutput, Workflow


def echo(step_input: StepInput) -> StepOutput:
    return StepOutput(content=str(step_input.input))


db = SqliteDb(db_file="tmp/workflow-api.db")
workflow = Workflow(id="echo", name="Echo", db=db, steps=[Step(name="Echo", executor=echo)])
agent_os = AgentOS(db=db, workflows=[workflow])
app = agent_os.get_app()

if __name__ == "__main__":
    agent_os.serve(app="workflow_api:app", host="127.0.0.1", port=7777)

Choose a response mode

message is a required form field, including when it contains structured JSON. A JSON request body does not replace the form. stream defaults to true; background defaults to false.

For example, submit background work:

curl --fail-with-body http://127.0.0.1:7777/workflows/echo/runs \
  -H "Authorization: Bearer $OS_SECURITY_KEY" \
  -F 'message=Hello' -F 'background=true' -F 'stream=false'

A typical acceptance body is:

{"run_id":"returned-run-id","session_id":"returned-session-id","status":"PENDING"}

Retain both returned IDs and use Get Workflow Run. 202 means accepted, not completed. A terminal run can also be paused, cancelled or failed.

Both background modes require a workflow database (the AgentOS database can supply it); otherwise they return 400. Remote workflows do not support background submission. A database alone does not provide restart durability: this example uses a process task. Durable execution requires an active configured queue and an eligible submission. Factory, version-pinned, off-registry or non-serializable submissions can bypass that queue.

Streaming events

Workflow streams use events such as WorkflowStarted and WorkflowCompleted. An abbreviated initial frame is:

event: WorkflowStarted
data: {"event":"WorkflowStarted","workflow_id":"echo","run_id":"returned-run-id","session_id":"returned-session-id"}

The generated RunStarted example below is not the workflow's event name. After SSE headers are sent, a failure can arrive as WorkflowError inside an HTTP 200 response; inspect events. Use Resume Workflow Run Stream to reconnect to background streaming runs. Poll background non-streaming runs instead.

Structured input and factories

For a workflow with an execution input_schema, JSON-encode a matching object in message, for example -F 'message={"text":"Hello"}' when that schema requires text. Inspect workflow details first. Invalid input returns 400 in non-streaming execution or acceptance; streaming can report validation failure as an event.

factory_input is a separate JSON object for constructing a dynamic workflow. It does not replace execution input. version selects a stored component configuration; it does not change a registered code workflow. With authentication configured, explicit draft preview is restricted to the owner or an administrator and can return 404 when unavailable. Unauthenticated open instances allow privileged preview.

curl --request POST 'https://example.com/workflows/string/runs' \  --data-urlencode 'message=string'

null

{  "detail": "string",  "error_id": "string",  "error_type": "string"}

{  "detail": "string",  "error_id": "string",  "error_type": "string"}

{  "detail": "string",  "error_id": "string",  "error_type": "string"}

{  "detail": "string"}

{  "detail": "string",  "error_id": "string",  "error_type": "string"}