Workflows

---

​

Overview

A workflow is an async Python function that calls tasks and combines their results. Use workflows for jobs with multiple steps, such as preparing data, running inference, and writing results. The workflow orchestrator records progress and applies each task’s retry and timeout settings.

Deploy a workflow with chalkcompute.deploy_workflow_worker. Chalk packages the source file and its dependencies into an image and creates a scaling group to run it. Each task can declare its own image and Python dependencies. Chalk creates a separate scaling group for each distinct task image, so steps with different environments can run in the same workflow.

Workers start at zero replicas. Submitting a run wakes the workflow’s worker; calling a task with its own image wakes that task’s worker. Each group stays up while it has an open workflow and returns to zero after its idle delay.

---

​

Define and deploy a workflow

Use an environment with a workflow orchestrator enabled, and authenticate with chalk login.

Define tasks with @task and the orchestration function with @workflow. Tasks can be synchronous or asynchronous; workflow functions must be async def. Save this example as pipeline.py:

# /// script
# dependencies = ["chalkcompute", "chalkpy[workflows]"]
# ///
import chalkcompute
from chalk.workflows import task, workflow


@task
def score(rows: list[int]) -> list[float]:
    return [round(row * 0.1, 2) for row in rows]


@task
def summarize(batch_id: str, scores: list[float]) -> dict:
    return {
        "batch_id": batch_id,
        "rows": len(scores),
        "mean_score": sum(scores) / len(scores) if scores else None,
    }


@workflow
async def score_batch(batch_id: str, rows: list[int]) -> dict:
    scores = await score(rows)
    return await summarize(batch_id, scores)


if __name__ == "__main__":
    worker = chalkcompute.deploy_workflow_worker(name="wf-score-batch")
    print("task queue:", worker.task_queue)

Run the script to deploy the worker:

uv run pipeline.py

The worker serves score_batch on the queue chalkcompute-workflows.wf-score-batch. Deploying the same name again updates the existing group. Deployment does not start a workflow run.

Keep deployment and submission code inside if __name__ == "__main__":. Workers import the file to register its tasks and workflows, so code outside that block also runs on each worker.

Workflow code must be deterministic. Put network requests, file access, randomness, and wall-clock reads inside tasks. Use JSON-serializable workflow inputs and return values.

---

​

Submit a run

Once the worker is deployed, submit runs by name from another script or service. The caller needs Chalk credentials and chalkpy[workflows]; it does not need to import pipeline.py.

Save this as submit.py, using the task queue printed by the deployment:

# /// script
# dependencies = ["chalkpy[workflows]"]
# ///
from chalk.client import ChalkClient

run = ChalkClient().trigger_workflow(
    "score_batch",
    {"batch_id": "batch-001", "rows": list(range(8))},
    task_queue="chalkcompute-workflows.wf-score-batch",
    wait=False,
)
print("workflow id:", run.workflow_id)
print("run id:", run.run_id)

uv run submit.py

wait=False returns a handle once the run is accepted. The workflow continues after the submitting process exits. Pass wait=True to wait for completion and receive the workflow’s return value instead of a handle:

result = ChalkClient().trigger_workflow(
    "score_batch",
    {"batch_id": "batch-001", "rows": list(range(8))},
    task_queue="chalkcompute-workflows.wf-score-batch",
    wait=True,
)
print(result)
# {'batch_id': 'batch-001', 'rows': 8, 'mean_score': 0.35}

The first task may wait for a worker to start and pull its image when the group is at zero replicas.

If you already have the worker returned by deploy_workflow_worker, use worker.trigger(...) to submit to its queue:

run = worker.trigger(
    score_batch,
    {"batch_id": "batch-002", "rows": [2, 4, 6]},
)

worker.trigger also accepts a workflow name and defaults to wait=False. Pass wait=True to return the result.

---

​

Task images and dependencies

Set image on a task to give it a separate Python environment. For example, replace score in pipeline.py with a task that uses NumPy:

numpy_image = chalkcompute.Image.debian_slim("3.12").pip_install(["numpy"])


@task(image=numpy_image, retries=2, timeout="5m")
def score(rows: list[int]) -> list[float]:
    import numpy as np

    return np.round(np.asarray(rows) * 0.1, 2).tolist()

Run uv run pipeline.py again to deploy the change. Chalk creates an additional scaling group for the NumPy image. score runs there, while score_batch and summarize run on the main worker. Tasks with equal image definitions share a group, including when the images are separate Python objects. Tasks without an image use the main worker’s environment.

Keep imports for task-specific packages inside the task body. Every worker imports the whole source file; importing NumPy at module scope would require it on the main worker and the deploying machine too.

The default worker image installs the dependencies in the script’s # /// script block, including chalkpy[workflows]. Each task image also receives those shared dependencies and the source file. Put packages shared by all workers in that block, and packages used only by a task on its image. See Images for installing system packages and adding local files.

Inspect the groups through worker.scaling_group for the main worker and worker.task_workers for the mapping from task names to their groups. Groups remain deployed at zero replicas between runs; scaling down does not delete them.

---

​

Worker configuration

Pass worker settings to deploy_workflow_worker:

worker = chalkcompute.deploy_workflow_worker(
    source="pipeline.py",
    name="wf-score-batch",
    cpu="2",
    memory="4Gi",
    max_replicas=2,
    shutdown_delay_seconds=60,
)

source defaults to the running script. Without an explicit name, the worker name is derived from the file name, such as wf-pipeline.

The main worker and task image workers use the same CPU, memory, replica limit, shutdown delay, env, and secrets settings. The helper sets min_replicas=0 and configures each group to scale on its workflow task queue. An open workflow keeps its group running, including while it waits for a task or retry.

Pass image to customize the main worker’s image. With a custom main image, install the script’s dependencies and chalkpy[workflows] yourself; the helper adds the source file.

---

​

Retries and parallel tasks

Configure retries and timeouts on each @task. In the NumPy example, retries=2 permits two retries after the first failed attempt, and timeout="5m" limits each attempt to five minutes. The defaults are no retries and a ten-minute timeout per attempt. Use retry_delay, such as retry_delay="30s", to set the initial retry interval.

A retried task can execute more than once. Make external writes safe to repeat, for example by using a batch ID as an idempotency key.

Use .submit() inside a workflow to start independent tasks before waiting for their results:

import asyncio


@workflow
async def score_batches(batches: list[list[int]]) -> list[list[float]]:
    handles = [score.submit(rows) for rows in batches]
    return await asyncio.gather(*handles)

Each call uses the task’s image, retry policy, and timeout. Add the workflow to pipeline.py and redeploy the worker before submitting it by name.

---

​

Run locally

Call a workflow directly to test it in the current process:

import asyncio
from pipeline import score_batch

result = asyncio.run(score_batch("local-batch", [2, 4, 6]))

Direct calls use the in-process orchestrator. They do not deploy scaling groups or use task images, so install each task’s dependencies locally.

To execute locally with durable coordination from the environment’s orchestrator, use ChalkClient().run_workflow:

from chalk.client import ChalkClient

result = ChalkClient().run_workflow(
    score_batch,
    {"batch_id": "local-batch", "rows": [2, 4, 6]},
)

This call blocks until the workflow returns. Its tasks execute in the calling process and ignore image; keep that process running for the duration of the workflow.