Quick Start

Install

[dependencies]
a3s-flow = "0.3"
tokio = { version = "1", features = ["full"] }
serde_json = "1"

pip install a3s-flow

npm install @a3s-lab/flow

Run a Workflow

use a3s_flow::{FlowEngine, NodeRegistry, ExecutionState};
use serde_json::json;
use std::collections::HashMap;

#[tokio::main]
async fn main() -> a3s_flow::Result<()> {
    let engine = FlowEngine::new(NodeRegistry::with_defaults());

    let definition = json!({
        "nodes": [
            { "id": "start", "type": "noop" },
            { "id": "process", "type": "noop" },
            { "id": "end", "type": "noop" }
        ],
        "edges": [
            { "source": "start", "target": "process" },
            { "source": "process", "target": "end" }
        ]
    });

    let id = engine.start(&definition, HashMap::new()).await?;

    match engine.state(id).await? {
        ExecutionState::Completed(result) => {
            println!("✓ Workflow completed");
            println!("Outputs: {:#?}", result.outputs);
        }
        other => println!("State: {}", other.as_str()),
    }

    Ok(())
}

import asyncio
from a3s_flow import FlowEngine

async def main():
    engine = FlowEngine()

    definition = {
        "nodes": [
            {"id": "start", "type": "noop"},
            {"id": "process", "type": "noop"},
            {"id": "end", "type": "noop"}
        ],
        "edges": [
            {"source": "start", "target": "process"},
            {"source": "process", "target": "end"}
        ]
    }

    execution_id = await engine.start(definition, {})
    state = await engine.state(execution_id)

    if state["status"] == "completed":
        print("✓ Workflow completed")
        print(f"Outputs: {state['result']['outputs']}")

asyncio.run(main())

import { FlowEngine } from '@a3s-lab/flow';

const engine = new FlowEngine();

const definition = {
  nodes: [
    { id: 'start', type: 'noop' },
    { id: 'process', type: 'noop' },
    { id: 'end', type: 'noop' }
  ],
  edges: [
    { source: 'start', target: 'process' },
    { source: 'process', target: 'end' }
  ]
};

const executionId = await engine.start(definition, {});
const state = await engine.state(executionId);

if (state.status === 'completed') {
  console.log('✓ Workflow completed');
  console.log('Outputs:', state.result.outputs);
}

Lifecycle Control

// Start execution
let id = engine.start(&definition, variables).await?;

// Pause at next wave boundary
engine.pause(id).await?;

// Resume execution
engine.resume(id).await?;

// Terminate execution
engine.terminate(id).await?;

// Query current state
let state = engine.state(id).await?;

HTTP Request Example

{
  "nodes": [
    {
      "id": "fetch",
      "type": "http-request",
      "data": {
        "url": "https://api.github.com/repos/A3S-Lab/Flow",
        "method": "GET",
        "headers": {
          "User-Agent": "a3s-flow"
        }
      }
    },
    {
      "id": "check",
      "type": "if-else",
      "data": {
        "cases": [
          {
            "id": "success",
            "conditions": [
              {
                "from": "fetch",
                "path": "status",
                "op": "eq",
                "value": 200
              }
            ]
          }
        ]
      }
    }
  ],
  "edges": [
    { "source": "fetch", "target": "check" }
  ]
}

Next Steps

SDK Reference — Complete API documentation for all three SDKs
Node Types — Built-in node types and configuration
Custom Nodes — Implement your own node types
Examples — Real-world workflow examples

Quick Start

Quick Start

Install

Run a Workflow

Lifecycle Control

HTTP Request Example

Next Steps

On this page