Workflow Resources

1. Concepts and Identifiers

What Is a Resource? A workflow resource is a workflow-scoped configuration object (connection settings, uploaded files, or ephemeral file state) that nodes can reuse at run time. Credentials and shared settings live in one place instead of being duplicated on every node.

You will encounter two main identifiers:

2. Resource Catalog

All types below are registered in SandsFlow.

3. Set Up Resource

In the workflow graph editor, open Workflow Resources from the left sidebar: click the + (plus) on the right side of the Resources row (next to the resource count).

That opens the Workflow Resources dialog, where you set Resource ID, Resource Type, any extra fields for the chosen type, and Add Resource.

Selecting a Resource Type updates the Resource Info panel with that resource’s metadata: a short description and the Resource Functions you can use with that resource. Expand a function to see its parameters and return type.

4. Per-Type Reference

For each type: Configuration lists config_schema fields shown in the editor. Runtime Functions are the methods exposed to node code (metadata functions); at execution time they are invoked on a wrapper that injects internal context. Each function row lists Description, Params, and Returns.

4.1 temporary_file — Temporary File

Description: Creates and manages temporary files for the duration of a workflow run. Files are tracked by a file_id (UUID). Typical uses: scratch output, intermediate artifacts, and workflow file parameters (the engine registers uploads against the matching temporary_file resource before nodes run). For how the SandsBytes node uses temporary_file for HTTP file parameters (resource id vs file_id), see SandsBytes workflow node — user guide.

Configuration (config_schema): None (empty). No extra fields in the resource form beyond Resource ID and Resource Type.

Runtime Functions:

Behavior: duplicate filenames in the same temporary resource raise FileExistsError. create writes an empty file (text-mode touch); use open(..., "wb") (or other modes) when you need binary or overwrite semantics.

Example:

def main(args, resources):
    tmp = resources.get("scratch")
    file_id = tmp.create("hello.txt")
    with tmp.open(file_id, mode="w") as f:
        f.write("hello")
    with tmp.open(file_id, mode="r") as f:
        content = f.read()
    return {"file_id": file_id, "content": content, "listed": tmp.list()}

4.2 persistent_file — Persistent File

Description: Stores one or more files at design time (uploads in the resource editor). At run time you address each file by file_id (UUID) stored in config.files (the id field on each file entry). You may also create empty files at run time; they are appended to the in-memory config.files list for that execution (whether that survives beyond the run depends on how your deployment persists workflow definitions).

Configuration (config_schema):

Runtime Functions:

Example:

def main(args, resources):
    docs = resources.get("policy_pdfs")
    file_id = "your-file-id-from-resource-config"
    path    = docs.get_path(file_id)
    with docs.open(file_id, mode="rb") as f:
        data = f.read()
    return {"path": path, "size": len(data)}

4.3 database — Database (PostgreSQL/MySQL)

Description: Connect to PostgreSQL or MySQL databases. Execute queries, fetch results, and manage transactions.

Configuration (config_schema):

Runtime Functions:

Example:

def main(args, resources):
    db = resources.get("orders_db")
    db.connect()
    try:
        rows = db.fetch_all(
            "SELECT id, name FROM customers WHERE active = %s",
            [True],
        )
        return {"customer_count": len(rows), "sample": rows[:5]}
    finally:
        db.close()

4.4 mongodb — MongoDB

Description: Connect to MongoDB databases. Insert, find, update, and delete documents.

Configuration (config_schema):

Runtime Functions:

Example:

def main(args, resources):
    mongo = resources.get("app_mongo")
    mongo.connect()
    try:
        inserted_id = mongo.insert_one("items", {"name": "demo", "count": 1})
        doc         = mongo.find_one("items", {"name": "demo"})
        return {"inserted_id": inserted_id, "doc": doc}
    finally:
        mongo.close()

4.5 redis — Redis

Description: Connect to Redis servers. Get, set, and manage key-value pairs.

Configuration (config_schema):

Runtime Functions:

Example:

def main(args, resources):
    cache = resources.get("my_redis")
    cache.connect()
    try:
        cache.set("last_run", str(args.get("run_key", "")))
        value = cache.get("some_key")
        return {"cached": value}
    finally:
        cache.close()

4.6 s3 — Amazon S3

Description: Connect to Amazon S3. Upload, download, and manage objects in buckets (optional custom endpoint for S3-compatible services).

Configuration (config_schema):

Runtime Functions:

Example:

def main(args, resources):
    store = resources.get("my_s3")
    store.connect()
    store.upload("my-bucket", "workflow/output.txt", "hello", content_type="text/plain")
    body = store.download("my-bucket", "workflow/output.txt")
    return {"body": body}

4.7 azure_blob — Azure Blob Storage

Description: Connect to Azure Blob Storage. Upload, download, and manage blobs in containers (default container in config).

Configuration (config_schema):

Runtime Functions:

Example:

def main(args, resources):
    blob = resources.get("azure_reports")
    blob.connect()
    blob.upload("daily-report.json", '{"ok": true}', content_type="application/json")
    text = blob.download("daily-report.json")
    return {"text": text}

4.8 rabbitmq — RabbitMQ

Description: Connect to RabbitMQ brokers. Publish and consume messages from queues.

Configuration (config_schema):

Runtime Functions:

Example:

def main(args, resources):
    mq = resources.get("my_amqp")
    mq.connect()
    try:
        mq.declare_queue("workflow_jobs", durable=True)
        mq.publish("", "workflow_jobs", '{"task": "ping"}')
        msg = mq.consume("workflow_jobs", auto_ack=True)
        return {"message": msg}
    finally:
        mq.close()

4.9 kafka — Apache Kafka

Description: Connect to Apache Kafka brokers. Produce and consume messages from topics.

Configuration (config_schema):

Runtime Functions:

Example:

def main(args, resources):
    k = resources.get("my_kafka")
    k.connect()
    try:
        k.produce("events", "hello-from-workflow", key="run-1")
        batch = k.consume("events", "python_code_group", timeout=1)
        return {"messages": batch[:5]}
    finally:
        k.close()

4.10 minio — MinIO

Description: Connect to MinIO object storage. Upload, download, and manage objects in buckets (S3-compatible API).

Configuration (config_schema):

Runtime Functions:

Example:

def main(args, resources):
    store = resources.get("minio_store")
    store.connect()
    store.upload("my-bucket", "runs/output.bin", b"\x00\x01", content_type="application/octet-stream")
    body = store.download("my-bucket", "runs/output.bin")
    return {"body": body}

4.11 nextcloud — Nextcloud

Description: Connect to Nextcloud servers. Upload, download, and manage files via WebDAV.

Configuration (config_schema):

Runtime Functions:

Example:

def main(args, resources):
    nc = resources.get("team_nextcloud")
    nc.connect()
    nc.upload("/Workflows/hello.txt", "hello")
    content = nc.download("/Workflows/hello.txt")
    names   = nc.list_files("/Workflows/")
    return {"content": content, "names": names}

5. Workflow Parameters and File Uploads

If the workflow params schema includes a field with type: file, then:

1. The field’s id must match a resource_id on the workflow’s resources list.

2. That resource must have resource_type temporary_file.

def main(args, resources):
    res = resources.get("temp_file")

    return args

At run start, uploaded files for that parameter are written into the corresponding temporary file resource state (via TemporaryFileResource.create then open(..., "wb") in run_workflow), and the run params entry for that field becomes a list of file_id values (UUID strings).

If the resource is missing or not temporary_file, run_workflow raises ValueError with an explicit message.