flux-client-sdk

The official TypeScript/JavaScript SDK for the Flux Orchestration Platform.

It allows you to define distributed jobs, build complex dependency graphs (DAGs), and schedule Cron tasks purely using code.

📦 Installation

npm i @manthankhawse/flux-client

🚀 Initialization

Connect to your Flux API endpoint (usually localhost during development).

const { Flux } = require('@manthankhawse/flux-client');

const client = new Flux('http://localhost:3000');

⚡ Standalone Jobs

Use createJob to define a reusable task definition, then call invoke to trigger it immediately.

1. Node.js Jobs

Node.js jobs behave like Serverless Functions. You provide a native Javascript function, and the SDK serializes it to be run inside the container.

Requirements:

• Must pass an async function.
• Return value is saved as job output.

const mathJob = client.createJob({
    name: 'node-calc',
    runtime: 'node:18',
    packages: ['lodash'], // Auto-install npm packages
    maxRetries: 3,
    
    // This function is serialized and sent to the worker
    handler: async (payload: any) => {
        const _ = require('lodash'); // Import installed packages dynamically
        console.log(`Processing numbers: ${payload.a}, ${payload.b}`);
        
        return { 
            sum: payload.a + payload.b,
            random: _.random(1, 100)
        };
    }
});

// Trigger execution
await mathJob.invoke({ a: 10, b: 20 });

2. Python Jobs

Python jobs are defined as string templates. You MUST define a function named handler(payload).

const pythonJob = client.createJob({
    name: 'python-analyser',
    runtime: 'python:3.9',
    packages: ['pandas'], // Auto-install pip packages
    
    handler: `
import json
import pandas as pd

def handler(payload):
    print(f"🐍 Analyzing user: {payload.get('username')}")
    
    # Return value is captured as JSON output
    return {
        "status": "processed",
        "score": 98.5
    }
`
});

await pythonJob.invoke({ username: "alice" });

3. Bash Jobs

Bash jobs run as raw shell scripts. The payload is accessible via the $PAYLOAD environment variable or by parsing payload.json.

const bashJob = client.createJob({
    name: 'system-check',
    runtime: 'bash',
    handler: `
echo "🔥 Starting System Check..."
echo "Current Dir: $(pwd)"

# Access input data
TARGET=$(cat payload.json | grep -o '"target":"[^"]*' | cut -d'"' -f4)
echo "Pinging $TARGET..."

sleep 2
echo "✅ Done"
`
});

await bashJob.invoke({ target: "database-prod" });

🔗 Workflows (DAGs)

Workflows allow you to chain jobs together. Dependent jobs (children) only execute after their parents successfully complete. The parent's return value is passed to the child's context.

const flow = client.workflow("daily-etl-pipeline");

// Step 1: Scrape Data
flow.addStep({
    id: 'step-1',
    runtime: 'node:18',
    packages: ['axios'],
    handler: async (payload) => {
        const axios = require('axios');
        const res = await axios.get('[https://api.coindesk.com/v1/bpi/currentprice.json](https://api.coindesk.com/v1/bpi/currentprice.json)');
        return { price: res.data.bpi.USD.rate_float };
    }
});

// Step 2: Analyze (Runs AFTER Step 1)
flow.addStep({
    id: 'step-2',
    runtime: 'python:3.9',
    dependsOn: ['step-1'], // <--- Define Dependency
    handler: `
def handler(payload):
    # 'context' contains outputs from previous steps
    price = payload['context']['step-1']['price']
    
    print(f"📉 Bitcoin Price: {price}")
    return { "buy_signal": price < 50000 }
`
});

// Submit the DAG
await flow.commit();

⏰ Cron Scheduling

You can attach a schedule to any workflow using standard Cron syntax.

const flow = client.workflow("hourly-cleanup");

flow.addStep({
    id: 'clean-tmp',
    runtime: 'bash',
    handler: 'rm -rf /tmp/*.log'
});

// Run every hour
flow.schedule("0 * * * *");

await flow.commit();

🧩 API Reference

createJob(options)

Creates a standalone job definition.

| Option | Type | Description | | --- | --- | --- | | name | string | Human readable name for the dashboard. | | runtime | 'node:18' | 'python:3.9' | 'bash' | Execution environment. | | packages | string[] | List of NPM/Pip packages to install. | | handler | Function | string | The source code to execute. | | maxRetries | number | Attempts allowed on failure (Default: 3). | | runAt | Date | Schedule execution for a specific future time. |

workflow(id)

Starts a workflow builder session.

• .addStep(options): Adds a node to the graph. Accepts same options as createJob, plus:

• id (Required): Unique string ID for this step.

• dependsOn: Array of step IDs that must finish before this one starts.

• .schedule(cronExpression): Sets a recurring schedule (e.g., * * * * *).

• .commit(): Uploads the workflow blueprint to the Flux Orchestrator.