fluid-workflow-engine-sdk

Python SDK for the Coredge Fluid workflow engine: remote step workers (gRPC), workflow definition building (YAML, workflow/v1 + workflow/v2 flow control), workflow/trigger registration, and execution management.

pip install fluid-workflow-engine-sdk            # core
pip install "fluid-workflow-engine-sdk[fastapi]" # + FastAPI integration
pip install "fluid-workflow-engine-sdk[flask]"   # + Flask integration

import fluid_workflow_engine_sdk

Migrating from workflow-engine-sdk (≤ 0.2.1)

The distribution was renamed workflow-engine-sdk → fluid-workflow-engine-sdk and the import package workflow_engine_sdk → fluid_workflow_engine_sdk in 0.3.0.

• import workflow_engine_sdk still works via a deprecated shim (emits DeprecationWarning); update imports at your convenience.
• Replace the old requirement line — never install both distributions in one environment. Both own the workflow_engine_sdk/ path; pip will silently clobber files and uninstalling either breaks the other.

- workflow_engine_sdk==0.2.1
+ fluid-workflow-engine-sdk==0.3.0

Worker quick start

A worker hosts step functions, registers them with the engine, and heartbeats. Steps are plain callables; declare only the keyword args you need — inputs, workflow_id, workflow_name, step_name, is_compensation, retry_count, and the optional signature-gated extras log, auth, and scope_path.

from fluid_workflow_engine_sdk import WorkerClient, RetryableError

worker = WorkerClient(
    service_name="my-svc",
    engine_address="workflow-engine:50052",
    worker_host="my-svc",
    grpc_port=50055,
)

@worker.step("createThing", description="Creates a thing", default_timeout="30s", max_retries=3)
def create_thing(inputs, log=None, auth=None, **_):
    if log:
        log("INFO", "creating", name=inputs["name"])
    if not_ready():
        raise RetryableError("dependency not ready")   # engine retries per policy
    return {"id": "thing-123"}                          # step outputs

@worker.step("deleteThing", supports_compensation=True)
def delete_thing(inputs, is_compensation, **_):
    ...
    return {}

worker.start()      # or: app = FastAPI(lifespan=worker.lifespan)

Anything logged with the stdlib logging module inside a step body is also streamed to the engine's event log (disable with WorkerClient(..., tee_logging=False)). Caveat: threads spawned inside a step body are not captured — pass log explicitly there.

Steps may be async def — each invocation runs on a fresh event loop on the worker's thread pool (asyncio.run), so log/auth and the logging tee work unchanged. Don't cache loop-bound resources (e.g. a module-level httpx.AsyncClient) across invocations; create them inside the step.

Useful constructor extras: grpc_port=0 binds an OS-assigned port (the bound port is what gets registered — required under pre-fork servers), and wait_for_engine_s=30 retries engine registration with backoff at startup. Introspection: worker.worker_id, worker.is_running, worker.bound_port, worker.active_executions, worker.step_metadata().

FastAPI integration

from fastapi import FastAPI
from fluid_workflow_engine_sdk.contrib.fastapi import worker_lifespan, worker_router

app = FastAPI(lifespan=worker_lifespan(worker))       # starts/stops the worker
app.include_router(worker_router(worker))             # GET /fluid/healthz, /fluid/steps

worker_lifespan runs the blocking start/stop off the event loop and composes with an existing lifespan: worker_lifespan(worker, inner=app_lifespan) (worker start → inner enter → serve → inner exit → worker stop).

Flask integration

from flask import Flask
from fluid_workflow_engine_sdk.contrib.flask import FlaskWorker

app = Flask(__name__)
FlaskWorker(worker, app)   # starts the worker now, stops it atexit,
                           # mounts /fluid/healthz + /fluid/steps

Under a pre-fork server (Gunicorn): construct with grpc_port=0, call FlaskWorker(worker).init_app(app, start=False) at import, and start each fork's worker in a post_fork hook — app.extensions["fluid_worker"].start(). Never start in the master under --preload (gRPC servers/channels don't survive fork()). See examples/gunicorn.conf.py. The Flask dev-server reloader imports the app twice — run with use_reloader=False.

Step routers

StepRouter decouples step declaration from the worker instance, like FastAPI's APIRouter — feature modules own their steps, main assembles:

from fluid_workflow_engine_sdk import StepRouter

billing = StepRouter(prefix="billing")   # advertised as "billing.<name>"

@billing.step("charge")
async def charge(inputs, **_):
    return {"chargeId": "..."}

worker.include_router(billing)           # ValueError on duplicate step names

Routers nest (router.include_router(other)) and can carry workflows (router.workflow(defn)), which register when the including worker starts.

Workflow auto-registration

worker.workflow(wf)                # WorkflowDefinition, YAML str, or bytes
worker.start()                     # registers steps, then pushes workflows

Definitions are pushed right after worker registration (source_service=service_name); a rejected definition raises EngineError and aborts startup. Steps whose function is served by this worker get executionMode: grpc defaulted in automatically (the engine dispatches remotely only on an explicit non-local mode); builtins and other services' functions are left untouched.

Configuration via environment

from fluid_workflow_engine_sdk import WorkerClient, WorkerSettings

settings = WorkerSettings.from_env()   # FLUID_SERVICE_NAME, FLUID_ENGINE_ADDRESS,
                                       # FLUID_WORKER_HOST, FLUID_GRPC_PORT,
                                       # FLUID_MAX_WORKERS, FLUID_TEE_LOGGING,
                                       # FLUID_WAIT_FOR_ENGINE_S
worker = WorkerClient.from_settings(settings)

Keyword overrides win over the environment; the legacy WORKFLOW_ENGINE_ADDRESS / SERVICE_HOST names are honored as deprecated fallbacks.

Building workflow definitions

from fluid_workflow_engine_sdk import WorkflowDefinition, Step, RetryPolicy

wf = (
    WorkflowDefinition("provision-fleet")
    .api_version("workflow/v2")            # required for flow-control operators
    .input("regions", type="list", required=True)
    .input("env", type="string", default="staging")
    .step(
        Step("deployAll")
        .foreach("inputs.regions", as_="region", parallel=True, max_concurrency=4)
        .body(
            Step("provision", function="createVpc")
            .input("region", "{{ loop.region }}")
            .retry(RetryPolicy(max_retries=3, initial_backoff="1s"))
            .compensation("deleteVpc", inputs={"region": "{{ loop.region }}"}),
        )
    )
    .step(
        Step("notify")
        .if_("inputs.env == 'prod'")
        .body(Step("page", function="pageOncall"))
        .else_(Step("slack", function="notifySlack"))
        .depends_on("deployAll")
    )
    .output("done", "{{ steps.deployAll.outputs.completed }}")
)
print(wf.to_yaml())

while_/until (polling loops, max_iterations mandatory) and switch/case/default are also available. Step inputs support the object form for optional values: Step(...).input("size", "{{ inputs.size }}", required=False, default="m5.large").

Engine client

from fluid_workflow_engine_sdk import WorkflowEngineClient, TriggerSpec

with WorkflowEngineClient("http://engine:50051", engine_grpc="engine:50052") as client:
    client.register_workflow(wf, replace=True, source_service="my-svc")
    res = client.start_workflow(
        "provision-fleet",
        {"regions": ["us-east-1"]},
        started_by="user:ashok",
        tenant="acme", domain="default", project="demo",
    )
    detail = client.get_execution(res.workflow_id)

    client.register_trigger(
        TriggerSpec(
            name="on-vm-delete",
            resource_type="compute",
            event_type="deleted",
            workflow_name="cleanup-vm",
            input_mappings={"vmName": "event.resource_name"},
        ),
        source_service="my-svc",
        replace=True,
    )

Also available: unregister_workflow, list_workflow_definitions (gRPC registry view), resume_from_step(workflow_id, step_name), unregister_trigger, list_triggers, plus the HTTP admin surface (list_workflows, get_workflow, list_executions, get_execution, cancel_execution).

Examples

See examples/ for runnable scripts: basic worker, FastAPI (plain + router/settings/composed-lifespan), Flask (+ Gunicorn config), workflow auto-registration, definition building with flow control, register-and-start, and trigger registration. Framework guides: docs/PYTHON_FASTAPI_WORKER.md and docs/PYTHON_FLASK_WORKER.md.

Development

pip install -e ".[dev]"
make gen-stubs         # regenerate gRPC stubs from ../api/workflow/workflow_service.proto
make test              # unit + offline integration tests (FakeEngine, no Docker)
make test-integration  # just the offline FakeEngine integration tier
make e2e-up            # MongoDB via docker compose (needs `make build` at repo root)
make test-e2e          # real engine + Mongo end-to-end tier
make e2e-down

grpcio-tools is pinned so regenerated stubs keep the same protobuf gencode version as the committed ones (protobuf 6.31.x); if you bump it, bump the protobuf runtime floor in pyproject.toml to match the new gencode requirement.