a3s-flow

A3S Flow Python SDK

Python bindings for the A3S Flow workflow engine.

Installation

pip install a3s-flow

Quick Start

from a3s_flow import FlowEngine

# Create a flow engine
engine = FlowEngine()

# Define a simple workflow
definition = {
    "nodes": [
        {"id": "start", "type": "noop"},
        {"id": "process", "type": "noop"}
    ],
    "edges": [
        {"source": "start", "target": "process"}
    ]
}

# Start the flow
execution_id = engine.start(definition, {})

# Check the state
state = engine.state(execution_id)
print(f"Status: {state.status}")

# Pause and resume
engine.pause(execution_id)
engine.resume(execution_id)

# Terminate
engine.terminate(execution_id)

API Reference

FlowEngine

FlowEngine()

Create a new flow engine with default built-in nodes.

start(definition: dict, variables: dict = None) -> str

Start a new flow execution.

• definition: Flow definition with "nodes" and "edges"
• variables: Initial variables (optional)
• Returns: Execution ID as a string

pause(execution_id: str)

Pause a running flow execution.

resume(execution_id: str)

Resume a paused flow execution.

terminate(execution_id: str)

Terminate a running flow execution.

state(execution_id: str) -> ExecutionState

Get the current state of a flow execution.

node_types() -> list[str]

List all registered node types.

ExecutionState

• status: "running", "paused", "completed", "failed", or "terminated"
• result: FlowResult (only present when status is "completed")
• error: Error message (only present when status is "failed")

FlowResult

• execution_id: Unique ID for this execution run
• outputs: Per-node outputs, keyed by node ID
• completed_nodes: IDs of all nodes that were processed
• skipped_nodes: IDs of nodes that were skipped
• context: Final snapshot of the shared mutable context

Development

Build from source

cd sdk/python
pip install maturin
maturin develop

Run tests

pytest

License

MIT