Running Workflows - Noxus Documentation

Overview

Once you’ve built and saved a workflow, you can execute it with different inputs and monitor its progress. The Noxus SDK provides both synchronous and asynchronous methods for running workflows and handling results.

Basic Workflow Execution

Running a Workflow

from noxus_sdk.client import Client

client = Client(api_key="your_api_key_here")

# Get an existing workflow
workflow = client.workflows.get("workflow_id_here")

# Run with input data
run = workflow.run(body={
    "User Input": "What are the benefits of renewable energy?",
    "Analysis Type": "comprehensive"
})

# Wait for completion
result = run.wait(interval=2)  # Check every 2 seconds

print(f"Status: {result.status}")
print(f"Output: {result.output}")

Input Formats

The input format depends on how you configured your workflow nodes:

Node Labels

Use the label you assigned to input nodes:

# If you created an input node with label "User Question"
input_node = workflow_def.node("InputNode").config(
    label="User Question"
)

# Run with this input
run = workflow.run(body={
    "User Question": "How does photosynthesis work?"
})

Node IDs

Use the node’s unique ID:

# Get node ID from workflow definition
input_node = workflow_def.node("InputNode")
node_id = input_node.id

# Run with node ID
run = workflow.run(body={
    node_id: "Your input value here"
})

Node ID with Input Name

Use the format {node_id}::{input_name}:

# For input nodes, the default input name is "input"
run = workflow.run(body={
    f"{node_id}::input": "Your input value here"
})

Monitoring Execution

Synchronous Monitoring

Wait for workflow completion with custom intervals:

# Run workflow
run = workflow.run(body={"input": "test data"})

# Monitor with custom interval and timeout
try:
    result = run.wait(
        interval=5,      # Check every 5 seconds
        timeout=300      # Maximum wait time (5 minutes)
    )

    print(f"Completed in {result.execution_time}ms")
    print(f"Final output: {result.output}")

except TimeoutError:
    print("Workflow execution timed out")
    print(f"Current status: {run.status}")

Asynchronous Monitoring

For non-blocking execution monitoring:

import asyncio

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

    # Start workflow execution
    run = await workflow.arun(body={"input": "test data"})

    # Monitor asynchronously
    result = await run.a_wait(interval=2)

    return result

# Run async monitoring
result = asyncio.run(monitor_workflow_async())

Manual Status Checking

For more control over monitoring:

import time

# Start workflow
run = workflow.run(body={"input": "test data"})

# Manual monitoring loop
while True:
    # Refresh run status
    run = run.refresh()

    print(f"Status: {run.status}")

    if run.status in ["completed", "failed", "cancelled"]:
        break

    time.sleep(3)  # Wait 3 seconds before next check

# Handle final result
if run.status == "completed":
    print(f"Success! Output: {run.output}")
elif run.status == "failed":
    print(f"Failed: {run.error_message}")

Real-Time Streaming

Instead of polling for run completion, you can stream events in real time using Server-Sent Events (SSE). This gives you instant notifications as nodes execute and when the run finishes.

Streaming Run Events

# Create a run and stream events as they happen
run = workflow.run(body={"User Input": "Explain quantum computing"})

for event in run.stream():
    if event.type == "content":
        node_id = event.data.get("id")
        status = event.data.get("status")
        content = event.data.get("content", "")
        print(f"[{node_id}] {status}: {content}")
    elif event.type == "state":
        print(f"Progress: {event.data.get('status')}")

    if event.is_terminal:
        print(f"Run finished: {event.data['workflow_status']}")

# Refresh to get final output after streaming
run.refresh()
print(f"Output: {run.output}")

One-Liner: Run and Stream

Create a run and immediately stream its events in a single call:

for event in workflow.run_and_stream(body={"User Input": "Hello"}):
    if event.type == "content" and event.data.get("content"):
        print(event.data["content"], end="")
    if event.is_terminal:
        print(f"\nDone: {event.data['workflow_status']}")

Async Streaming

import asyncio

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

    async for event in workflow.arun_and_stream(body={"input": "test"}):
        if event.type == "content":
            print(event.data)
        if event.is_terminal:
            break

asyncio.run(stream_workflow())

RunEvent Properties

Each event yielded by stream() is a RunEvent object:

Property	Type	Description
`event.type`	`str`	`"content"` (node output) or `"state"` (progress)
`event.data`	`dict`	Event payload with node ID, status, content, etc.
`event.is_terminal`	`bool`	`True` when `workflow_status` is `"completed"` or `"failed"`
`event.redis_id`	`str`	Stream cursor — pass as `etag` to resume from this point

run.wait() now uses SSE streaming internally for instant completion detection. If the SSE endpoint is unavailable, it automatically falls back to polling.

Webhook Callbacks

For event-driven integrations, you can provide a callback_url when creating a run. The platform will POST the run result to your URL when the run reaches a terminal state (completed, failed, or stopped).

Setting Up a Callback

run = workflow.run(
    body={"User Input": "Analyze this data"},
    callback_url="https://your-server.com/webhook/noxus"
)

# No need to poll — your server will be notified
print(f"Run {run.id} started, webhook will fire on completion")

Webhook Payload

Your endpoint receives a POST with this JSON body:

{
  "run_id": "abc-123",
  "workflow_id": "wf-456",
  "status": "completed",
  "outputs": {
    "node_id::output": {
      "text": "The generated result...",
      "file": null
    }
  },
  "error": null,
  "started_at": "2025-01-15T10:30:00.000000",
  "completed_at": "2025-01-15T10:30:05.000000",
  "duration_ms": 5000
}

Webhook Behavior

Retries: Failed deliveries are retried up to 3 times with exponential backoff (1s, 5s, 15s)
Validation: The callback_url must be a valid HTTP(S) URL — invalid URLs are rejected at creation time
Non-blocking: Webhook delivery never affects run execution — if your endpoint is down, the run still completes normally
Terminal states: Webhooks fire on completed, failed, and stopped statuses

Handling Different Input Types

Text Inputs

# Simple text input
run = workflow.run(body={
    "Question": "Explain quantum computing",
    "Context": "For a general audience"
})

File Inputs

# Method 1: Using a public file URL (downloaded server-side, max 25MB).
# Must be publicly reachable — private/loopback/metadata hosts are rejected.
run = workflow.run(body={
    "Document": {"uri": "https://path/to/file.pdf", "name": "file.pdf"}
})

# Method 2: Using base64-encoded file content (data URI format)
import base64

with open("document.pdf", "rb") as file:
    file_content = file.read()
    base64_content = base64.b64encode(file_content).decode("utf-8")
    data_uri = f"data:application/pdf;base64,{base64_content}"

run = workflow.run(body={
    "Document": {"uri": data_uri, "name": "document.pdf"}
})

# Method 3: Reference a file already uploaded via POST /v1/file
run = workflow.run(body={
    "Document": {"uri": "spot://your-file-id", "name": "file.pdf"}
})

Multiple Inputs

# Workflow with multiple input nodes
run = workflow.run(body={
    "Primary Text": "The main content to analyze",
    "Analysis Focus": "sentiment and themes",
    "Output Format": "bullet points",
    "Max Length": "300 words"
})

Fixed vs Dynamic Inputs

# Only provide values for dynamic inputs
# Fixed inputs (configured with fixed_value=True) are automatically used
run = workflow.run(body={
    "User Question": "What is machine learning?"
    # Fixed inputs like system prompts are handled automatically
})

# Override fixed inputs if needed (if the workflow allows it)
run = workflow.run(body={
    "User Question": "What is machine learning?",
    "System Prompt": "You are a technical expert. Be precise and detailed."
})

Result Handling

Understanding Run Results

result = run.wait()

# Basic result information
print(f"Run ID: {result.id}")
print(f"Status: {result.status}")
print(f"Started: {result.created_at}")
print(f"Completed: {result.updated_at}")
print(f"Execution time: {result.execution_time}ms")

# Output data
print(f"Final output: {result.output}")

# Error information (if failed)
if result.status == "failed":
    print(f"Error: {result.error_message}")
    print(f"Error details: {result.error_details}")

Processing Output Data

# Output is typically a string, but can be structured data
output = result.output

if isinstance(output, str):
    # Text output
    print(f"Generated text: {output}")
elif isinstance(output, dict):
    # Structured output
    for key, value in output.items():
        print(f"{key}: {value}")
elif isinstance(output, list):
    # List output
    for i, item in enumerate(output):
        print(f"Item {i}: {item}")

Large or Truncated Text Outputs

When a text output exceeds 16,000 characters, the API does not return the full string inline. Instead, it truncates the value in the response, uploads the full content to file storage, and returns a TextContainer-shaped dict with a has_preview flag and a file reference:

{
    "text": "...first 16,000 characters of the output...",
    "has_preview": True,
    "file": {
        "id": "abc-123",
        "uri": "spot://abc-123",
        "name": "truncated_result.txt",
        "content_type": "text/plain",
        "size": 52314
    }
}

Detect this shape via has_preview and fetch the full content through the Files API:

result = run.wait()

for key, value in result.output.items():
    if isinstance(value, dict) and value.get("has_preview"):
        # Output was truncated — download the full content from storage
        file_id = value["file"]["id"]
        full_content = client.files.get(file_id).decode("utf-8")
        print(f"{key}: full length {len(full_content)} chars")
    else:
        # Output fit under the limit — use `text` (or the raw string) directly
        text = value.get("text") if isinstance(value, dict) else value
        print(f"{key}: {text}")

has_preview only appears on outputs that were actually truncated. Outputs that fit under the 16,000-character limit are returned either as a plain string or as {"text": "..."} without a file reference, so always guard on has_preview before attempting to download.

Saving Results

import json
from datetime import datetime

# Save result to file
result_data = {
    "workflow_id": workflow.id,
    "run_id": result.id,
    "timestamp": datetime.now().isoformat(),
    "status": result.status,
    "output": result.output,
    "execution_time": result.execution_time
}

with open(f"workflow_result_{result.id}.json", "w") as f:
    json.dump(result_data, f, indent=2)

print(f"Result saved to workflow_result_{result.id}.json")

Batch Processing

Running Multiple Workflows

import asyncio

async def run_multiple_workflows():
    client = Client(api_key="your_api_key_here")

    # Get workflows
    workflow1 = await client.workflows.aget("workflow_id_1")
    workflow2 = await client.workflows.aget("workflow_id_2")

    # Start multiple runs concurrently
    run1 = await workflow1.arun(body={"input": "data for workflow 1"})
    run2 = await workflow2.arun(body={"input": "data for workflow 2"})

    # Wait for all to complete
    result1, result2 = await asyncio.gather(
        run1.a_wait(),
        run2.a_wait()
    )

    return result1, result2

# Execute batch processing
results = asyncio.run(run_multiple_workflows())

Processing Multiple Inputs

async def process_multiple_inputs(workflow, inputs):
    """Process multiple inputs through the same workflow"""

    # Start all runs
    runs = []
    for input_data in inputs:
        run = await workflow.arun(body=input_data)
        runs.append(run)

    # Wait for all to complete
    results = []
    for run in runs:
        result = await run.a_wait()
        results.append(result)

    return results

# Usage
inputs = [
    {"Question": "What is AI?"},
    {"Question": "How does machine learning work?"},
    {"Question": "What are neural networks?"}
]

workflow = client.workflows.get("qa_workflow_id")
results = asyncio.run(process_multiple_inputs(workflow, inputs))

for i, result in enumerate(results):
    print(f"Answer {i+1}: {result.output}")

Error Handling

Handling Execution Errors

def run_workflow_safely(workflow, input_data):
    """Run workflow with comprehensive error handling"""

    try:
        # Start the workflow
        run = workflow.run(body=input_data)

        # Wait for completion with timeout
        result = run.wait(interval=2, timeout=600)  # 10 minute timeout

        if result.status == "completed":
            return {"success": True, "output": result.output}
        elif result.status == "failed":
            return {
                "success": False,
                "error": result.error_message,
                "details": result.error_details
            }
        else:
            return {"success": False, "error": f"Unexpected status: {result.status}"}

    except TimeoutError:
        return {"success": False, "error": "Workflow execution timed out"}
    except Exception as e:
        return {"success": False, "error": f"Execution error: {str(e)}"}

# Usage
result = run_workflow_safely(workflow, {"input": "test data"})

if result["success"]:
    print(f"Success: {result['output']}")
else:
    print(f"Error: {result['error']}")

Retry Logic

import time
import random

def run_workflow_with_retry(workflow, input_data, max_retries=3):
    """Run workflow with exponential backoff retry"""

    for attempt in range(max_retries):
        try:
            run = workflow.run(body=input_data)
            result = run.wait(interval=2, timeout=300)

            if result.status == "completed":
                return result
            elif result.status == "failed":
                if attempt == max_retries - 1:
                    raise Exception(f"Workflow failed: {result.error_message}")

                # Wait before retry with exponential backoff
                wait_time = (2 ** attempt) + random.uniform(0, 1)
                print(f"Attempt {attempt + 1} failed, retrying in {wait_time:.1f}s...")
                time.sleep(wait_time)

        except Exception as e:
            if attempt == max_retries - 1:
                raise

            wait_time = (2 ** attempt) + random.uniform(0, 1)
            print(f"Error on attempt {attempt + 1}: {e}")
            print(f"Retrying in {wait_time:.1f}s...")
            time.sleep(wait_time)

# Usage
try:
    result = run_workflow_with_retry(workflow, {"input": "test data"})
    print(f"Success after retries: {result.output}")
except Exception as e:
    print(f"Failed after all retries: {e}")

Performance Optimization

Efficient Polling

def adaptive_wait(run, initial_interval=1, max_interval=30, timeout=600):
    """Wait for workflow completion with adaptive polling"""

    start_time = time.time()
    interval = initial_interval

    while True:
        run = run.refresh()

        if run.status in ["completed", "failed", "cancelled"]:
            return run

        # Check timeout
        if time.time() - start_time > timeout:
            raise TimeoutError("Workflow execution timed out")

        # Adaptive interval - increase polling interval over time
        time.sleep(interval)
        interval = min(interval * 1.5, max_interval)

# Usage
run = workflow.run(body={"input": "test data"})
result = adaptive_wait(run)

Concurrent Execution with Limits

import asyncio
from asyncio import Semaphore

async def run_workflows_with_limit(workflows_and_inputs, max_concurrent=5):
    """Run multiple workflows with concurrency limit"""

    semaphore = Semaphore(max_concurrent)

    async def run_single(workflow, input_data):
        async with semaphore:
            run = await workflow.arun(body=input_data)
            return await run.a_wait()

    # Create tasks for all workflows
    tasks = [
        run_single(workflow, input_data)
        for workflow, input_data in workflows_and_inputs
    ]

    # Execute with concurrency limit
    results = await asyncio.gather(*tasks, return_exceptions=True)

    return results

# Usage
workflows_and_inputs = [
    (workflow1, {"input": "data1"}),
    (workflow2, {"input": "data2"}),
    (workflow3, {"input": "data3"}),
    # ... more workflows
]

results = asyncio.run(run_workflows_with_limit(workflows_and_inputs))

Best Practices

Input Validation

Validate inputs before running workflows:

def validate_workflow_input(input_data, required_fields):
    """Validate workflow input data"""
    missing_fields = []
    
    for field in required_fields:
        if field not in input_data or not input_data[field]:
            missing_fields.append(field)
    
    if missing_fields:
        raise ValueError(f"Missing required fields: {missing_fields}")
    
    return True

# Usage
required_fields = ["Question", "Context"]
validate_workflow_input(input_data, required_fields)

Resource Management

Manage resources efficiently:

# Use context managers for resource cleanup
async def process_workflows():
    client = Client(api_key="your_api_key")
    try:
        # Process workflows
        results = await run_multiple_workflows()
        return results
    finally:
        # Cleanup resources if needed
        pass

Logging and Monitoring

Implement comprehensive logging:

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def run_workflow_with_logging(workflow, input_data):
    logger.info(f"Starting workflow {workflow.id}")
    logger.info(f"Input data: {input_data}")
    
    start_time = time.time()
    run = workflow.run(body=input_data)
    
    logger.info(f"Workflow started, run ID: {run.id}")
    
    result = run.wait()
    execution_time = time.time() - start_time
    
    logger.info(f"Workflow completed in {execution_time:.2f}s")
    logger.info(f"Status: {result.status}")
    
    if result.status == "failed":
        logger.error(f"Workflow failed: {result.error_message}")
    
    return result

Next Steps

Node Types

Learn about all available node types and their capabilities

Workflow Examples

Explore complete workflow examples for common use cases

Advanced Patterns

Master advanced workflow design and execution patterns

API Reference

Detailed API reference for workflow execution

​Overview

​Basic Workflow Execution

​Running a Workflow

​Input Formats

​Monitoring Execution

​Synchronous Monitoring

​Asynchronous Monitoring

​Manual Status Checking

​Real-Time Streaming

​Streaming Run Events

​One-Liner: Run and Stream

​Async Streaming

​RunEvent Properties

​Webhook Callbacks

​Setting Up a Callback

​Webhook Payload

​Webhook Behavior

​Handling Different Input Types

​Text Inputs

​File Inputs

​Multiple Inputs

​Fixed vs Dynamic Inputs

​Result Handling

​Understanding Run Results

​Processing Output Data

​Large or Truncated Text Outputs

​Saving Results

​Batch Processing

​Running Multiple Workflows

​Processing Multiple Inputs

​Error Handling

​Handling Execution Errors

​Retry Logic

​Performance Optimization

​Efficient Polling

​Concurrent Execution with Limits

​Best Practices

​Next Steps

Node Types

Workflow Examples

Advanced Patterns

API Reference

Overview

Basic Workflow Execution

Running a Workflow

Input Formats

Monitoring Execution

Synchronous Monitoring

Asynchronous Monitoring

Manual Status Checking

Real-Time Streaming

Streaming Run Events

One-Liner: Run and Stream

Async Streaming

RunEvent Properties

Webhook Callbacks

Setting Up a Callback

Webhook Payload

Webhook Behavior

Handling Different Input Types

Text Inputs

File Inputs

Multiple Inputs

Fixed vs Dynamic Inputs

Result Handling

Understanding Run Results

Processing Output Data

Large or Truncated Text Outputs

Saving Results

Batch Processing

Running Multiple Workflows

Processing Multiple Inputs

Error Handling

Handling Execution Errors

Retry Logic

Performance Optimization

Efficient Polling

Concurrent Execution with Limits

Best Practices

Next Steps