# Dapr Workflow: code that picks up where it left off

- URL: https://baeke.info/2026/11/03/dapr-workflow-code-that-picks-up-where-it-left-off/
- Author: Geert Baeke
- Published: November 3, 2026
- Topics: ai, dapr, workflows, python

Lots of the work I see in projects is a chain of steps: fetch a document, convert it, call a model, write the result somewhere. You write it as a background job that reads from a queue, and it works fine until the process restarts halfway through a 300-page PDF or a model call times out. Before long you're writing status tables, retry loops and checks for "did I already do this step?", and that code is bigger than the job itself.

Dapr Workflow takes that part off your hands. You write the steps as ordinary code, and the workflow engine records what has happened, so a workflow that crashes continues from the last completed step. To try it on something real, I built a small document extraction service with three Dapr services. You upload a document and a JSON schema, and a workflow turns the document into Markdown and extracts the fields with OpenAI structured outputs. The code is on GitHub: https://github.com/gbaeke/dapr-workflow

This is the pipeline we'll walk through, with numbers that match the sections further down:

![The upload service stores a document and publishes a message, the process service starts a durable workflow with six activities that pass blob names between them, and Dapr components connect it all to blob storage and Redis](https://baeke.info/_astro/pipeline.Dk8Yuurd_Z1sPo25.webp)

The repo has a second pipeline that runs a LangChain deep agent as a durable workflow. That one gets its own post. Here we stick to the basics.

## What is Dapr?

[Dapr](https://dapr.io) (Distributed Application Runtime) is an open source CNCF project that runs next to your app as a sidecar. Your code calls the sidecar over HTTP or gRPC for things most services need: publish a message, save state, read a file from storage. The sidecar talks to the actual infrastructure through components, small YAML files that say "pub/sub is Redis" or "this binding is Azure Blob Storage". Swap the YAML and the same code works against Service Bus or Kafka.

Each of those capabilities is a building block. Workflow is one of them, and it has been stable since [Dapr 1.15](https://blog.dapr.io/posts/2025/02/27/dapr-v1.15-is-now-available/). There are SDKs for Python, .NET, Java, Go and JavaScript. I used Python.

⚠️ **Important:** Azure Container Apps has Dapr built in, so it looks like the obvious place to run this. Workflow isn't among the Dapr APIs that Container Apps supports, though, and the docs list the workflow SDK packages as incompatible ([Dapr in Azure Container Apps](https://learn.microsoft.com/azure/container-apps/dapr-overview#compatible-sdks)). For Dapr Workflow on Azure, use Kubernetes: AKS has a Dapr extension, and Learn has a walkthrough for [workflows on AKS](https://learn.microsoft.com/azure/aks/dapr-workflow).

## What is Dapr Workflow?

A workflow is a function that calls activities. An activity is a plain function that does the actual work: call an API, read a blob, ask a model. The workflow decides the order, passes outputs from one activity to the next and can run activities in parallel.

The engine runs that function in an unusual way. It keeps an append-only history of everything that happened: workflow started, activity scheduled, activity completed with this result. Each time the workflow waits for an activity, the engine saves that history to a state store. Then the function stops; nothing waits in memory. Every time an activity finishes, for every workflow, the engine runs your workflow function again **from the top**. For activities that already completed, it returns the recorded result instead of calling them again, until the code reaches a step it hasn't done yet. That's called [replay](https://docs.dapr.io/developing-applications/building-blocks/workflow/workflow-features-concepts/), and it's why a crash costs you very little: on restart, the engine replays the history and continues with the first step that has no result.

The workflow code runs in your app, and the engine runs in the Dapr sidecar. When the app starts, the workflow SDK opens a gRPC stream to the sidecar, and the sidecar sends work over it: run this workflow, run this activity. The app needs no inbound port for that. The top half of this diagram shows those parts, and the bottom half shows a replay:

![Your app holds the workflow function and the activities, the Dapr sidecar schedules the work over a gRPC stream and handles retries and timers, and the state store keeps the history and status per instance. Below, a first run completes make_greeting before the process dies, and after the restart the function runs from the top, gets make_greeting's result from the history and only schedules send_greeting](https://baeke.info/_astro/workflow.CCpwws1-_1dA1eS.webp)

A minimal workflow in Python looks like this:

```python
from dapr.ext.workflow import DaprWorkflowContext, WorkflowRuntime

wfr = WorkflowRuntime()

@wfr.workflow(name="hello_wf")
def hello_wf(ctx: DaprWorkflowContext, name: str):
    greeting = yield ctx.call_activity(make_greeting, input=name)
    yield ctx.call_activity(send_greeting, input=greeting)
    return greeting

@wfr.activity(name="make_greeting")
def make_greeting(ctx, name: str) -> str:
    return f"Hello {name}"

# ...
```

Every `yield` is a point where the workflow can stop and later be replayed. If the process dies after `make_greeting` completed, the next run gets `greeting` from the history and only calls `send_greeting`.

Replay comes with one rule you have to respect: the workflow function must be **deterministic**. Given the same history it must make the same calls in the same order. So no `datetime.now()`, no random values or new UUIDs, and no direct I/O in the workflow function. Anything that touches the outside world goes into an activity. The context gives you replay-safe alternatives, such as `ctx.current_utc_datetime` and durable timers.

## Why would you want it?

You stop writing the plumbing around long-running jobs. Out of the box you get:

- resume after a crash, from the last completed activity
- retry policies per activity (attempts, backoff, timeout), and they survive restarts too
- fan-out/fan-in: start activities in parallel and wait for all of them with `when_all`
- durable timers that can wait minutes or months without holding a process
- external events, so a workflow can wait for a human approval or a callback
- a status API per workflow instance, with its input, output and failure details

The Dapr docs describe the usual shapes as [workflow patterns](https://docs.dapr.io/developing-applications/building-blocks/workflow/workflow-patterns/): task chaining, fan-out/fan-in, async HTTP APIs, monitors, waiting for external events and compensation (the saga pattern). The demo uses the first three.

If you know Azure Durable Functions or Temporal, this will feel familiar. The engine is built on [durabletask-go](https://docs.dapr.io/developing-applications/building-blocks/workflow/workflow-architecture/), and the programming model is the same idea of an orchestrator, activities and replay. The difference is that it runs anywhere Dapr runs, next to any app.

## What about AI agents?

An agent is a long-running, multi-step process too, and its steps are the flaky kind. A model call can take a minute, hit a rate limit or fail. A tool call might touch a system that's down. An agent run of 20 steps that dies on step 18 and starts over from zero costs time and tokens. And with agents doing longer tasks, a human approval in the middle of a run becomes normal, which is exactly what external events are for.

There are two ways to bring workflows and LLMs together:

- You define the steps in code and make each LLM call an activity. The model does the thinking inside a step, the workflow decides the order. The extract pipeline in this post works like that.
- You let the agent decide the steps and run each model turn and tool call as an activity. The agent loop becomes the workflow. That's the analyze pipeline in the repo, and the topic of the next post.

Both give you the same durability. The first is predictable and easy to reason about, the second is flexible.

⚠️ **Important:** this post uses the first approach, so there is no agent deciding what to do. I wrote the steps and their order in code. Two of them, `extract_structured` and `merge_extractions`, call a model, and the model only fills in the JSON schema; it never picks the next step. An agent that plans its own steps is the topic of the next post.

## The demo

The repo has three FastAPI services, each with a Dapr sidecar. `dapr run -f dapr.yaml` starts all of them:

- upload (port 8100) accepts the document and starts a pipeline
- process (port 8101) runs the extract workflow
- analyze (port 8102) runs a LangChain deep agent as a workflow (next post). Deep Agents support comes ready-made in [`diagrid`](https://pypi.org/project/diagrid/), Diagrid's extension SDK for Dapr, which wraps about ten agent frameworks in Dapr Workflows.

Next to the services there's a `components` folder with three Dapr components: `blobstore` (an Azure Blob Storage binding pointed at Azurite), `pubsub` and `statestore` (both Redis, the container `dapr init` starts for you). The state store holds the workflow history.

⚠️ **Important:** this is a local demo. The components use the Azurite development key, the services allow all CORS origins and there's no authentication. Keep it on your laptop. 😊

## 1. The upload service

The upload service doesn't know what happens to the document. It writes the file to blob storage through the binding and publishes a message. Here is the core of `POST /upload`:

```python
job_id = uuid.uuid4().hex
doc_blob = f"{job_id}/original{ext}"

with DaprClient() as client:
    client.invoke_binding(
        binding_name="blobstore",
        operation="create",
        data=base64.b64encode(content),
        binding_metadata={"blobName": doc_blob},
    )
    client.publish_event(
        pubsub_name="pubsub",
        topic_name="process-topic",
        data=json.dumps({"job_id": job_id, "doc_blob": doc_blob, "file_name": file.filename, "schema": schema_obj}),
        data_content_type="application/json",
    )
# ...
```

There's no storage SDK and no Redis client in this code. Both are Dapr components, so moving from Azurite to a real storage account, or from Redis to Azure Service Bus, is a YAML change.

## 2. Starting the workflow

The process service subscribes to `process-topic`. For each message it schedules a new workflow and acks the message:

```python
@dapr_app.subscribe(pubsub="pubsub", topic="process-topic")
def on_process_request(event_data=Body()):
    payload = event_data.get("data")
    # ...
    DaprWorkflowClient().schedule_new_workflow(
        workflow=doc_processing_wf, input=payload, instance_id=payload["job_id"]
    )
    # ...
    return {"success": True}
```

The workflow's instance id is the job id from the message. That's the id you use later to ask for the status, so the client only needs the job id it got back from the upload.

The handler also returns quickly. It doesn't wait for the workflow to finish; the work happens in the workflow engine, and the message is done.

## 3. The workflow

Before the code, here is the flow. Follow the branch after `plan_chunks`: most documents take the short path on top, a big one fans out into parallel extractions and a merge:

![The doc_processing_wf flow: a message on process-topic starts retrieve_document, convert_to_markdown and plan_chunks. With one chunk, a single extract_structured runs; with two or more, extract_structured runs once per chunk in parallel and merge_extractions combines the results. Both paths end in write_results and the workflow completes](https://baeke.info/_astro/flow.BbyD_0MB_Z1grv7e.webp)

The workflow function reads like that diagram. This is `doc_processing_wf` from `process/workflow.py`, a bit trimmed:

```python
retry_policy = RetryPolicy(
    first_retry_interval=timedelta(seconds=2),
    max_number_of_attempts=3,
    backoff_coefficient=2,
    max_retry_interval=timedelta(seconds=30),
    retry_timeout=timedelta(minutes=5),
)

@wfr.workflow(name="doc_processing_wf")
def doc_processing_wf(ctx: DaprWorkflowContext, wf_input: dict):
    job_id = wf_input["job_id"]
    if not ctx.is_replaying:
        logger.info("job %s: workflow started", job_id)

    meta = yield ctx.call_activity(retrieve_document, input={"doc_blob": wf_input["doc_blob"]}, retry_policy=retry_policy)
    markdown_blob = yield ctx.call_activity(convert_to_markdown, input={...}, retry_policy=retry_policy)
    chunk_blobs = yield ctx.call_activity(plan_chunks, input={...}, retry_policy=retry_policy)

    if len(chunk_blobs) == 1:
        extraction = yield ctx.call_activity(extract_structured, input={...}, retry_policy=retry_policy)
    else:
        # fan out: one extraction per chunk, all running in parallel
        tasks = [
            ctx.call_activity(extract_structured, input={"markdown_blob": chunk, "schema": wf_input["schema"]}, retry_policy=retry_policy)
            for chunk in chunk_blobs
        ]
        partials = yield when_all(tasks)
        extraction = yield ctx.call_activity(merge_extractions, input={"partials": partials, ...}, retry_policy=retry_policy)

    blobs = yield ctx.call_activity(write_results, input={"job_id": job_id, "extraction": extraction}, retry_policy=retry_policy)
    return {"job_id": job_id, "chunks": len(chunk_blobs), "markdown_blob": markdown_blob, **blobs}
```

The `if not ctx.is_replaying` check keeps the log clean. The function runs again from the top after every completed activity, so without the check you'd see "workflow started" once per step.

The `if` on the number of chunks is fine in a workflow. It depends on the result of `plan_chunks`, which comes from the history during a replay, so every replay takes the same branch. You can use all the logic you want, as long as it only depends on inputs and activity results.

The `else` branch is fan-out/fan-in. Calling `ctx.call_activity` without `yield` only creates a task. `when_all` starts them together and waits until all of them are done. Most documents fit in one model call, but the full text of the Odyssey (about 305k tokens) doesn't. It gets split into two chunks, both are extracted in parallel and a merge step combines the partial results. That run takes about two and a half minutes. 📜

The workflow sets the retry policy, so the activity code has no retry loops. When `extract_structured` raises, because of a 429 or a timeout, the engine waits 2 seconds and tries again, then 4 seconds, up to three attempts. Those retries run on durable timers, so they also survive a restart of the process.

## Activities pass blob names

The activities are ordinary functions with a decorator. This is the conversion step:

```python
@wfr.activity(name="convert_to_markdown")
def convert_to_markdown(ctx: WorkflowActivityContext, act_input: dict) -> str:
    content = _get_blob(act_input["doc_blob"])
    # ... write to a temp file
    markdown = MarkItDown().convert(tmp_path).text_content
    if not markdown or not markdown.strip():
        raise ValueError("document conversion produced no text")

    markdown_blob = f"{act_input['job_id']}/converted.md"
    _create_blob(markdown_blob, markdown.encode(), "text/markdown")
    return markdown_blob
```

It returns the name of the blob where it wrote the Markdown. I learned that one the hard way. Activity inputs and outputs end up in the workflow history, and the history travels between the sidecar and the app over gRPC. My first version passed the document content around as base64. A 1.7 MB PDF pushed the history past the 4 MB message limit and the worker got stuck. Since then, each activity reads its input from storage and writes its output to storage, and the workflow only passes names around. The history stays small whether the document is 2 pages or 600.

⚠️ **Important:** keep workflow inputs, outputs and activity results small. The limit comes from the sidecar's maximum body size (4 MiB by default), and the [docs](https://docs.dapr.io/developing-applications/building-blocks/workflow/workflow-features-concepts/) say a workflow whose history passes 95% of it stalls. Store large data somewhere else and pass a reference.

The extraction activity is where the model comes in. It reads the Markdown and calls OpenAI with the uploaded schema as `response_format`:

```python
@wfr.activity(name="extract_structured")
def extract_structured(ctx: WorkflowActivityContext, act_input: dict) -> dict:
    markdown = _get_blob(act_input["markdown_blob"]).decode()
    return _call_openai(
        "Extract information from the user's document according to the provided JSON schema. ...",
        markdown,
        act_input["schema"],
    )
```

`_call_openai` first tries strict structured outputs. Strict mode wants `additionalProperties: false` on every object and every property in `required`, and a schema somebody uploads often doesn't follow those rules. When OpenAI rejects the schema, the function retries in non-strict mode. A refusal or an empty answer raises an exception, and then the workflow's retry policy takes over.

One more thing about activities: Dapr runs them **at least once**. If the process crashes after an activity did its work but before the result was saved, the activity runs again. Writing the same blob twice is harmless, so every activity in this demo is safe to repeat. Sending an email or charging a card would need more care.

## 4. Checking the status

Because the engine stores everything, the status endpoint is a thin wrapper:

```python
@app.get("/status/{job_id}")
def status(job_id: str):
    state = DaprWorkflowClient().get_workflow_state(instance_id=job_id, fetch_payloads=True)
    # ...
    return {
        "status": state.runtime_status.name,
        "output": json.loads(state.serialized_output) if state.serialized_output else None,
        "failure": state.failure_details.message if state.failure_details else None,
        # ...
    }
```

That's the async HTTP API pattern: the upload returns a job id right away, and the client polls until the status says `COMPLETED`.

## Running it

You need Docker, the Dapr CLI, uv and an OpenAI API key. The README has the full steps. In short:

```bash
dapr init                      # one time: Redis, placement and scheduler containers
cp process/.env.example process/.env   # add OPENAI_API_KEY
docker compose up -d           # Azurite
dapr run -f dapr.yaml          # all services with their sidecars
```

Then upload the sample invoice with its schema and ask for the status:

```bash
curl -F "file=@sample/invoice.html" -F "schema=<sample/schema.json" http://localhost:8100/upload
# {"job_id": "...", "pipeline": "extract", "status_url": "http://localhost:8101/status/..."}

curl http://localhost:8101/status/<job_id>
# {"status": "COMPLETED", "output": {"markdown_blob": "...", "result_blob": "..."}, ...}
```

The repo also has a small test UI in `ui/index.html`. It fires up to 500 uploads in parallel and shows each job with its status and links to the extracted JSON and the Markdown. In this screenshot, look at the status column: most jobs completed, one failed and one has no workflow state:

![The test UI with the upload panel on the left and a documents table on the right listing 21 jobs, 19 completed, 1 failed and 1 unknown, with links to the JSON and Markdown of each job](https://baeke.info/_astro/test-ui.DQUiYIfe_2t4FY.webp)

To see what the workflows are doing, I used the free [Diagrid Dev Dashboard](https://docs.diagrid.io/develop/local-development/dev-dashboard/). It finds the apps that `dapr run` started, without configuration, and reads workflow data from the local Redis state store. Under Workflows you get each `doc_processing_wf` instance with its event history, the inputs and outputs of every activity, and buttons to terminate or purge an instance. The Applications view shows the services and their sidecars:

![The Diagrid Dev Dashboard Applications view listing the process and upload apps as healthy, with their app, HTTP and gRPC ports](https://baeke.info/_astro/dev-dashboard.Bmh5g3DK_23OEEP.webp)

A good test is to stop `dapr run` while a big document is being processed and start it again. The workflow continues with the activity that was running, and the activities that completed before the stop don't run again.

## What you don't get

- Determinism is on you. The SDK doesn't stop you from calling `datetime.now()` or a database in the workflow function. It breaks during replay, sometimes much later.
- Changing a workflow while instances are running is tricky. Replay expects the same sequence of calls, so adding or reordering activities can break instances that started with the old code. Plan for versioning before you deploy changes.
- History stays in the state store after a workflow completes, until you purge it. With many jobs a day, add a purge.
- Not every state store works. You need one that supports actors (Redis, PostgreSQL, SQL Server and others); the [workflow overview](https://docs.dapr.io/developing-applications/building-blocks/workflow/workflow-overview/) lists limits for Cosmos DB and DynamoDB.
- You run Dapr yourself: sidecars, the placement and scheduler services, and the state store. Locally `dapr init` does it all. In production that means Kubernetes, for example AKS with the Dapr extension.

## Wrapping up

Dapr Workflow lets you write a long-running job as a plain function and leave crashes and retries to the engine. Pass references between activities, make activities safe to repeat and keep the workflow function deterministic, and you get a lot of reliability for very little code.

The code is at https://github.com/gbaeke/dapr-workflow. In the next post we'll put a LangChain deep agent inside a workflow, where the agent decides the steps and every model turn and tool call becomes a durable activity. If you spot any errors or have questions, feel free to reach out!