Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.noxus.ai/llms.txt

Use this file to discover all available pages before exploring further.

Every recipe below targets the same workflow run lifecycle:
  • Async create — POST /v1/workflows/{workflow_id}/runs returns immediately with a run id.
  • Sync create — POST /v1/workflows/{workflow_id}/runs/sync blocks server-side and returns the output.
  • Poll — GET /v1/workflows/{workflow_id}/runs/{run_id}.
  • Stream — GET /v1/runs/{run_id}/events (Server-Sent Events).
Replace workflow_id and your_api_key. The body’s input keys are your workflow’s input-node labels (or IDs).

​
Create a run and wait for the result

Simplest option — blocks until the run finishes. Best for short, interactive flows. Avoid it for long-running or highly parallel workloads, where it ties up a connection for the whole run.
# pip install noxus-sdk
from noxus_sdk.client import Client

client = Client(api_key="your_api_key")
workflow = client.workflows.get(workflow_id="workflow_id")

run = workflow.run(body={"User Question": "What is machine learning?"})
result = run.wait(interval=5)  # polls/streams under the hood until terminal
print(result.output)

​
Create a run and poll

Create asynchronously (returns a run id immediately), then check the run’s status at your own cadence. The right default for long-running flows or when launching many runs in parallel — your process never blocks on a single run.
import time
from noxus_sdk.client import Client

client = Client(api_key="your_api_key")
workflow = client.workflows.get(workflow_id="workflow_id")

run = workflow.run(body={"User Question": "What is machine learning?"})
while run.refresh().status not in ("completed", "failed"):
    print(f"status={run.status} progress={run.progress}%")
    time.sleep(2)
print(run.output)

​
Create a run and stream events (SSE)

Streams progress over Server-Sent Events as each node finishes. Best for live UIs and long flows where you want incremental feedback instead of one final payload. Each event has a type and a data payload; the stream ends when the run reaches a terminal state.
from noxus_sdk.client import Client

client = Client(api_key="your_api_key")
workflow = client.workflows.get(workflow_id="workflow_id")

# Creates the run and yields events until it reaches a terminal state
for event in workflow.run_and_stream(body={"User Question": "What is machine learning?"}):
    print(event.type, event.data)

​
Async / await with the SDK

Every SDK method has an a-prefixed coroutine variant (aget, arun, a_wait, arefresh, astream, arun_and_stream). Use them inside an async application so the event loop is never blocked.
Python SDK
import asyncio
from noxus_sdk.client import Client


async def main():
    client = Client(api_key="your_api_key")
    workflow = await client.workflows.aget(workflow_id="workflow_id")

    run = await workflow.arun(body={"User Question": "What is machine learning?"})

    # Wait for the result...
    result = await run.a_wait(interval=5)
    print(result.output)

    # ...or stream events instead:
    # async for event in run.astream():
    #     print(event.type, event.data)


asyncio.run(main())

​
Sending files as inputs

If a workflow has a File, Image, or Audio input node, pass that input as an object with a uri (and ideally a name) instead of a string. Three shapes are accepted:
  • Public URL — {"uri": "https://…", "name": "…"}. Noxus downloads it server-side and stores a copy. The URL must be publicly reachable; private, loopback, and cloud-metadata addresses are rejected, and downloads are capped at 25 MB.
  • Base64 data URI — {"uri": "data:<mime>;base64,<data>", "name": "…"}. Noxus decodes and stores it. Best for files you can’t expose over a URL.
  • Existing Noxus file — {"uri": "spot://<file-id>", "name": "…"}, where the id comes from a prior POST /v1/file upload.
The key is the input node’s label (here, "Document"); everything else in the body works exactly like the recipes above (wait, poll, or stream). 
from noxus_sdk.client import Client

client = Client(api_key="your_api_key")
workflow = client.workflows.get(workflow_id="workflow_id")

# Public URL
run = workflow.run(body={
    "Document": {"uri": "https://example.com/report.pdf", "name": "report.pdf"},
})

# …or base64 (e.g. a local file)
import base64
with open("report.pdf", "rb") as f:
    data = base64.b64encode(f.read()).decode()
run = workflow.run(body={
    "Document": {"uri": f"data:application/pdf;base64,{data}", "name": "report.pdf"},
})

print(run.wait().output)
Outputs that produce files come back as {"text": ..., "file": {...}} objects on the relevant output key — read the file metadata (including its URL) from the run output.

​
Webhooks (fire-and-forget)

Pass a callback_url when creating a run and Noxus will POST the result to it when the run reaches a terminal state — no polling or streaming needed. 
run = workflow.run(
    body={"User Question": "What is machine learning?"},
    callback_url="https://your-app.example.com/noxus/webhook",
)
See Create Async Run for the full webhook payload, retry, and timeout behaviour.