Workflows: run & get results - Noxus Documentation

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)

import requests

resp = requests.post(
    "https://backend.noxus.ai/v1/workflows/workflow_id/runs/sync",
    json={"input": {"User Question": "What is machine learning?"}},
    headers={"X-API-KEY": "your_api_key", "Content-Type": "application/json"},
)
resp.raise_for_status()
print(resp.json())  # the run output

const resp = await fetch(
  "https://backend.noxus.ai/v1/workflows/workflow_id/runs/sync",
  {
    method: "POST",
    headers: { "X-API-KEY": "your_api_key", "Content-Type": "application/json" },
    body: JSON.stringify({ input: { "User Question": "What is machine learning?" } }),
  }
);
console.log(await resp.json());

curl -X POST "https://backend.noxus.ai/v1/workflows/workflow_id/runs/sync" \
  -H "X-API-KEY: your_api_key" \
  -H "Content-Type: application/json" \
  -d '{"input": {"User Question": "What is machine learning?"}}'

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)

import time
import requests

base = "https://backend.noxus.ai/v1/workflows/workflow_id"
headers = {"X-API-KEY": "your_api_key", "Content-Type": "application/json"}

# 1. Create the run
run = requests.post(
    f"{base}/runs",
    json={"input": {"User Question": "What is machine learning?"}},
    headers=headers,
).json()
run_id = run["id"]

# 2. Poll until terminal
while True:
    run = requests.get(f"{base}/runs/{run_id}", headers=headers).json()
    if run["status"] in ("completed", "failed"):
        break
    time.sleep(2)
print(run.get("output"))

const base = "https://backend.noxus.ai/v1/workflows/workflow_id";
const headers = { "X-API-KEY": "your_api_key", "Content-Type": "application/json" };
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

// 1. Create the run
const created = await fetch(`${base}/runs`, {
  method: "POST",
  headers,
  body: JSON.stringify({ input: { "User Question": "What is machine learning?" } }),
}).then((r) => r.json());

// 2. Poll until terminal
let run;
do {
  await sleep(2000);
  run = await fetch(`${base}/runs/${created.id}`, { headers }).then((r) => r.json());
  console.log(`status=${run.status} progress=${run.progress}%`);
} while (!["completed", "failed"].includes(run.status));
console.log(run.output);

#!/bin/bash
base="https://backend.noxus.ai/v1/workflows/workflow_id"

# 1. Create the run
run_id=$(curl -s -X POST "$base/runs" \
  -H "X-API-KEY: your_api_key" -H "Content-Type: application/json" \
  -d '{"input": {"User Question": "What is machine learning?"}}' | jq -r '.id')

# 2. Poll until terminal
while true; do
  run=$(curl -s "$base/runs/$run_id" -H "X-API-KEY: your_api_key")
  status=$(echo "$run" | jq -r '.status')
  echo "status=$status"
  [ "$status" = "completed" ] || [ "$status" = "failed" ] && break
  sleep 2
done
echo "$run" | jq '.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)

import json
import requests

base = "https://backend.noxus.ai"
headers = {"X-API-KEY": "your_api_key", "Content-Type": "application/json"}

run = requests.post(
    f"{base}/v1/workflows/workflow_id/runs",
    json={"input": {"User Question": "What is machine learning?"}},
    headers=headers,
).json()

with requests.get(
    f"{base}/v1/runs/{run['id']}/events",
    headers={"X-API-KEY": "your_api_key"},
    stream=True,
) as resp:
    for line in resp.iter_lines():
        if line and line.startswith(b"data:"):
            event = json.loads(line[len(b"data:"):])
            print(event.get("type"), event.get("data"))

const base = "https://backend.noxus.ai";

const created = await fetch(`${base}/v1/workflows/workflow_id/runs`, {
  method: "POST",
  headers: { "X-API-KEY": "your_api_key", "Content-Type": "application/json" },
  body: JSON.stringify({ input: { "User Question": "What is machine learning?" } }),
}).then((r) => r.json());

const resp = await fetch(`${base}/v1/runs/${created.id}/events`, {
  headers: { "X-API-KEY": "your_api_key" },
});
const reader = resp.body.getReader();
const decoder = new TextDecoder();
while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  process.stdout.write(decoder.decode(value)); // raw SSE frames
}

#!/bin/bash
base="https://backend.noxus.ai"

run_id=$(curl -s -X POST "$base/v1/workflows/workflow_id/runs" \
  -H "X-API-KEY: your_api_key" -H "Content-Type: application/json" \
  -d '{"input": {"User Question": "What is machine learning?"}}' | jq -r '.id')

# -N disables curl's output buffering so events arrive live
curl -N "$base/v1/runs/$run_id/events" -H "X-API-KEY: your_api_key"

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)

import base64
import requests

headers = {"X-API-KEY": "your_api_key", "Content-Type": "application/json"}

# Public URL
requests.post(
    "https://backend.noxus.ai/v1/workflows/workflow_id/runs/sync",
    json={"input": {"Document": {"uri": "https://example.com/report.pdf", "name": "report.pdf"}}},
    headers=headers,
)

# …or base64
with open("report.pdf", "rb") as f:
    data = base64.b64encode(f.read()).decode()
requests.post(
    "https://backend.noxus.ai/v1/workflows/workflow_id/runs/sync",
    json={"input": {"Document": {"uri": f"data:application/pdf;base64,{data}", "name": "report.pdf"}}},
    headers=headers,
)

# Public URL
curl -X POST "https://backend.noxus.ai/v1/workflows/workflow_id/runs/sync" \
  -H "X-API-KEY: your_api_key" -H "Content-Type: application/json" \
  -d '{"input": {"Document": {"uri": "https://example.com/report.pdf", "name": "report.pdf"}}}'

# Base64 (build the data URI from a local file)
data=$(base64 -w0 report.pdf)
curl -X POST "https://backend.noxus.ai/v1/workflows/workflow_id/runs/sync" \
  -H "X-API-KEY: your_api_key" -H "Content-Type: application/json" \
  -d "{\"input\": {\"Document\": {\"uri\": \"data:application/pdf;base64,$data\", \"name\": \"report.pdf\"}}}"

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",
)

curl -X POST "https://backend.noxus.ai/v1/workflows/workflow_id/runs" \
  -H "X-API-KEY: your_api_key" -H "Content-Type: application/json" \
  -d '{
    "input": {"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.

​Create a run and wait for the result

​Create a run and poll

​Create a run and stream events (SSE)

​Async / await with the SDK

​Sending files as inputs

​Webhooks (fire-and-forget)

Create a run and wait for the result

Create a run and poll

Create a run and stream events (SSE)

Async / await with the SDK

Sending files as inputs

Webhooks (fire-and-forget)