flow-like-wasm-sdk (Python)

Python SDK for building Flow-Like WASM nodes using componentize-py. Write your node logic in plain Python — the SDK handles compilation to a WASM component via the WIT Component Model.

Install

pip install flow-like-wasm-sdk
# or with uv
uv add flow-like-wasm-sdk

For optional JSON schema support (typed pins with Pydantic):

pip install "flow-like-wasm-sdk[schema]"

Quick Start — Single Node

from flow_like_wasm_sdk import (
    NodeDefinition,
    PinDefinition,
    Context,
    ExecutionResult,
)


def get_definition() -> NodeDefinition:
    node = NodeDefinition(
        name="uppercase",
        friendly_name="Uppercase",
        description="Converts text to uppercase",
        category="Text/Transform",
    )
    node.add_pin(PinDefinition.input_exec("exec"))
    node.add_pin(PinDefinition.input_pin("text", "String", default=""))
    node.add_pin(PinDefinition.output_exec("exec_out"))
    node.add_pin(PinDefinition.output_pin("result", "String"))
    return node


def run(ctx: Context) -> ExecutionResult:
    text = ctx.get_string("text") or ""
    ctx.set_output("result", text.upper())
    return ctx.success("exec_out")

Quick Start — Node Package (multiple nodes)

from flow_like_wasm_sdk import NodeDefinition, PinDefinition, Context, ExecutionResult, PackageNodes

def define_add() -> NodeDefinition:
    node = NodeDefinition(name="add", friendly_name="Add", description="Adds two numbers", category="Math")
    node.add_pin(PinDefinition.input_exec("exec"))
    node.add_pin(PinDefinition.input_pin("a", "Float", default="0"))
    node.add_pin(PinDefinition.input_pin("b", "Float", default="0"))
    node.add_pin(PinDefinition.output_exec("exec_out"))
    node.add_pin(PinDefinition.output_pin("result", "Float"))
    return node

def run_add(ctx: Context) -> ExecutionResult:
    a = ctx.get_float("a") or 0.0
    b = ctx.get_float("b") or 0.0
    ctx.set_output("result", a + b)
    return ctx.success("exec_out")


pkg = PackageNodes()
pkg.add_node(define_add(), run_add)
# pkg.add_node(define_subtract(), run_subtract)

Testing with MockHostBridge

from flow_like_wasm_sdk import Context, ExecutionInput, MockHostBridge

def test_uppercase():
    host = MockHostBridge()
    ctx = Context(ExecutionInput(inputs={"text": '"hello"'}), host=host)
    result = run(ctx)
    assert result.outputs["result"] == '"HELLO"'
    assert result.exec_output == "exec_out"

Building to WASM

The recommended workflow uses componentize-py:

# Install componentize-py
pip install componentize-py

# Build WASM component
componentize-py \
  --wit-path path/to/flow-like.wit \
  componentize my_node \
  -o build/my_node.wasm

Or use the wasm-node-python template which includes the full build setup.

Publishing

uv build && uv publish

API Reference

NodeDefinition

NodeDefinition(
    name: str,
    friendly_name: str,
    description: str,
    category: str,
)

PinDefinition

Context

ExecutionResult

ExecutionResult(
    outputs: dict[str, str],   # JSON-encoded values
    exec_output: str | None,   # which exec pin to fire
    error: str | None,
)

pip install flow-like-wasm-sdk

Quick Example

from flow_like_wasm_sdk import (
    NodeDefinition,
    PinDefinition,
    Context,
    ExecutionResult,
    PackageNodes,
)


def get_definition() -> NodeDefinition:
    node = NodeDefinition("upper", "Uppercase", "Converts text to uppercase", "Text/Transform")
    node.add_pin(PinDefinition.input_exec("exec"))
    node.add_pin(PinDefinition.input_pin("text", "String", default=""))
    node.add_pin(PinDefinition.output_exec("exec_out"))
    node.add_pin(PinDefinition.output_pin("result", "String"))
    return node


def run(ctx: Context) -> ExecutionResult:
    text = ctx.get_string("text") or ""
    ctx.set_output("result", text.upper())
    return ctx.success()

Multi-node packages

from flow_like_wasm_sdk import PackageNodes

pkg = PackageNodes()
pkg.add_node(get_definition())
print(pkg.to_json())

Testing

Use MockHostBridge for local testing:

from flow_like_wasm_sdk import Context, ExecutionInput, MockHostBridge

host = MockHostBridge()
ctx = Context(ExecutionInput(inputs={"text": "hello"}), host=host)
result = run(ctx)
assert result.outputs["result"] == "HELLO"