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.

---

​

Wait for messages

A workflow can pause until someone sends it a message, such as a reviewer’s decision. Call wait_for_message with the message name and an optional timeout:

from datetime import timedelta

from chalk.workflows import WorkflowMessageTimeoutError, task, wait_for_message, workflow


@task
def apply_decision(account_id: str, approved: bool, reviewer: str) -> None:
    ...


@workflow
async def review_account(account_id: str) -> str:
    try:
        review = await wait_for_message("review", timeout=timedelta(days=2))
    except WorkflowMessageTimeoutError:
        return "expired"
    await apply_decision(account_id, review.payload["approved"], review.sender_id)
    return "approved" if review.payload["approved"] else "rejected"

Send the message from any script or service with Chalk credentials, using the workflow id of the run:

from chalk.client import ChalkClient

client = ChalkClient()
run = client.trigger_workflow(
    "review_account",
    {"account_id": "acct-1"},
    task_queue="chalkcompute-workflows.wf-score-batch",
    wait=False,
)
client.send_workflow_message(
    run.workflow_id,
    "review",
    {"approved": True, "comment": "Verified by phone"},
)

The workflow receives a WorkflowMessage with name, payload, sender_kind, sender_id, and sent_at. Chalk sets the sender from the caller’s credentials: sender_kind is "user" or "service_token", and sender_id is the user’s id or the service token’s id (not its client id). The caller cannot set the sender, so a workflow can use it to decide whose decisions to accept.

Messages are durable. A message sent before the workflow reaches wait_for_message is kept until the workflow asks for it, and messages with the same name arrive in the order they were sent. A workflow can wait for days; it stays open while it waits, which keeps its worker group running (see Worker configuration).

Sending a message requires the query.online permission, and each send is recorded in the audit log. Sending to a workflow that is not running raises an error. The payload must be JSON-serializable and at most 256 KiB.

---

​

Child workflows

A workflow can start another workflow as its child with start_child. The child runs as its own workflow on the same workers, with its own history. start_child returns a handle: await it for the child’s result, and call handle.send_message(...) to message the child.

Inside a workflow, send_message sends a message to another running workflow, which receives it with wait_for_message. In this example a supervisor starts an assistant, sends it questions, and waits for each answer. The assistant replies to whoever asked by using the message’s sender_id:

from datetime import timedelta

from chalk.workflows import send_message, start_child, task, wait_for_message, workflow


@task
def answer_question(question: str) -> str:
    ...


@workflow
async def assistant() -> int:
    answered = 0
    while True:
        ask = await wait_for_message("ask", timeout=timedelta(hours=1))
        if ask.payload.get("stop"):
            return answered
        answer = await answer_question(ask.payload["question"])
        await send_message(ask.sender_id, "answer", {"answer": answer})
        answered += 1


@workflow
async def supervisor(questions: list[str]) -> list[str]:
    helper = await start_child(assistant)
    answers = []
    for question in questions:
        await helper.send_message("ask", {"question": question})
        reply = await wait_for_message("answer", timeout=timedelta(minutes=30))
        answers.append(reply.payload["answer"])
    await helper.send_message("ask", {"stop": True})
    await helper
    return answers

Pass the child’s inputs as a mapping, such as start_child(score_batch, {"batch_id": "b-1", "rows": [1, 2]}), or pass a workflow name instead of the function. The child’s id defaults to one derived from the parent’s id and the child’s name; pass workflow_id to choose it. Starting a child with the id of a workflow that is still running fails.

A child’s lifetime is bounded by its parent’s. If the parent completes, fails, or is cancelled while a child is still running, the child is cancelled, and its finally blocks still run. Await every child before the parent returns. Children are not retried as a whole; configure retries on their tasks.

Inside a child, parent_workflow_id() returns the parent’s id, and current_workflow_id() returns the child’s own id, which other workflows can pass to send_message.

Messages sent with send_message are delivered by the workflow orchestrator, not through the Chalk API. Their sender_kind is "workflow" and their sender_id is the sending workflow’s id, as written by that workflow, so Chalk does not authenticate them the way it does messages from send_workflow_message.

The workflow’s page in the Chalk dashboard shows its lineage: the parent, its children, and the messages they sent each other.

Child workflows and send_message require chalkpy 2.167.0 or later.

---

​

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.

A direct call has no address that messages can be sent to. To test a workflow that calls wait_for_message, queue its messages with local_messages; they arrive with sender_kind "local":

from chalk.workflows import local_messages

with local_messages([("review", {"approved": True})]):
    result = asyncio.run(review_account("acct-1"))

start_child runs the child in the same process and cancels it if the parent returns first. send_message raises an error in a direct call.

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.