diff --git a/template/.env.jinja b/template/.env.jinja
index 5e39863..cf49985 100644
--- a/template/.env.jinja
+++ b/template/.env.jinja
@@ -1,6 +1,8 @@
 MISTRAL_API_KEY={{ mistral_api_key }}
 MISTRAL_API_URL=https://api.mistral.ai
 COLLECTION_NAME={{ collection_name }}
+# Required for workflow examples (make start-examples / make execute-ingestion)
+DEPLOYMENT_NAME={{ _copier_conf.dst_path.name }}
 WORKSPACE_ROOT=.
 # Optional: change only if ports 18080 / 19072 are already in use
 VESPA_QUERY_PORT=18080
diff --git a/template/Makefile.jinja b/template/Makefile.jinja
index 4064ec2..dac49be 100644
--- a/template/Makefile.jinja
+++ b/template/Makefile.jinja
@@ -1,4 +1,4 @@
-.PHONY: installdeps setup-vespa start-vespa stop-vespa reset-vespa migrate-vespa verify-vespa ingest search bruno generate-vespa-lock
+.PHONY: installdeps install-workflows setup-vespa start-vespa stop-vespa reset-vespa migrate-vespa verify-vespa ingest search bruno generate-vespa-lock start-examples execute-ingestion
 
 ifneq (,$(wildcard .env))
 include .env
@@ -65,3 +65,16 @@ generate-vespa-lock:
 	uv run mistral-vespa generate \
 		--app-dir src/vespa_app \
 		--path ./vespa.lock
+
+## Install optional workflows dependency (required for examples/workflows/)
+install-workflows:
+	uv sync --extra workflows
+
+## Start a worker that registers the example workflows (requires install-workflows)
+start-examples: install-workflows
+	uv run python -m examples.workflows.worker
+
+## Execute the ingestion workflow via the Mistral Workflows API
+## Usage: make execute-ingestion input='{"file_path": "sample_data/hello.txt", "collection_name": "mydocs"}'
+execute-ingestion: install-workflows
+	uv run python -m examples.workflows.start --workflow document-ingestion $(if $(input),--input '$(input)',--input '{"file_path":"sample_data/hello.txt"}')
diff --git a/template/pyproject.toml.jinja b/template/pyproject.toml.jinja
index 108ef9c..361ef8a 100644
--- a/template/pyproject.toml.jinja
+++ b/template/pyproject.toml.jinja
@@ -8,6 +8,12 @@ dependencies = [
     "python-dotenv>=0.9.9",
 ]
 
+[project.optional-dependencies]
+workflows = [
+    "mistralai-workflows[mistralai]>=3.0.0,<4",
+    "pydantic",
+]
+
 [dependency-groups]
 dev = [
     "ruff>=0.11",
@@ -16,6 +22,9 @@ dev = [
 
 [tool.uv]
 exclude-newer = "7 days"
+# Gemfury (UV_EXTRA_INDEX_URL) may list older versions of mistral-common /
+# mistralai-search-toolkit than PyPI. Allow uv to pick the best match across indexes.
+index-strategy = "unsafe-best-match"
 
 [tool.uv.exclude-newer-package]
 mistralai = false
@@ -28,4 +37,4 @@ requires = ["hatchling"]
 build-backend = "hatchling.build"
 
 [tool.hatch.build.targets.wheel]
-packages = ["src/entrypoints", "src/vespa_app"]
+packages = ["src/entrypoints", "src/vespa_app", "src/examples"]
diff --git a/template/src/examples/__init__.py b/template/src/examples/__init__.py
new file mode 100644
index 0000000..eba47ca
--- /dev/null
+++ b/template/src/examples/__init__.py
@@ -0,0 +1,7 @@
+"""Examples for the search-starter-app.
+
+Workflow-based examples live under ``examples/workflows/`` — see that folder's
+README for setup and usage.
+
+Other examples that do not use the workflows framework live directly here.
+"""
diff --git a/template/src/examples/workflows/README.md b/template/src/examples/workflows/README.md
new file mode 100644
index 0000000..e2c6712
--- /dev/null
+++ b/template/src/examples/workflows/README.md
@@ -0,0 +1,133 @@
+# Workflow Examples
+
+This folder contains examples that integrate the search-starter-app with the [Mistral Workflows](https://docs.mistral.ai/capabilities/workflows/) framework.
+
+Workflow-based examples live under `examples/workflows/`. Other examples that do not use the workflows framework (standalone scripts, direct API calls, etc.) live directly under `examples/`.
+
+## Available example
+
+| Example | Workflow name | Makefile target | Description |
+| --- | --- | --- | --- |
+| [search/](search/) | `document-ingestion` | `make execute-ingestion` | Ingestion workflow with small Temporal I/O |
+
+The example ingests local files into Vespa using the same Search Toolkit components as `make ingest`, wrapped in a durable workflow.
+
+## Prerequisites
+
+The workflows framework is an **optional** dependency — the core search project works without it.
+
+> Run these commands from your **generated project root** (the folder created by `copier copy`), not from the `search-starter-app` template repo itself.
+
+```bash
+make install-workflows   # uv sync --extra workflows
+make setup-vespa
+```
+
+Ensure these are set in your `.env` file:
+
+- `MISTRAL_API_KEY`
+- `DEPLOYMENT_NAME` — a stable identifier for this worker (defaults to your project name when generated via `copier copy`)
+
+## Workflow input
+
+```json
+{
+  "file_path": "sample_data/hello.txt",
+  "collection_name": "exampledocs"
+}
+```
+
+| Field | Required | Default | Description |
+| --- | --- | --- | --- |
+| `file_path` | yes | — | Path to a file or directory to ingest |
+| `collection_name` | no | `"exampledocs"` | Vespa collection name |
+
+## How to run
+
+> **Important:** execution commands call the Mistral Workflows API. The workflow must be registered first by a running worker. If you see `Workflow not found`, start the worker below and retry.
+
+**Terminal 1 — start the examples worker (leave this running):**
+
+```bash
+make start-examples
+```
+
+Wait until the worker is listening for execution requests.
+
+**Terminal 2 — trigger the workflow:**
+
+```bash
+make execute-ingestion
+make execute-ingestion input='{"file_path": "sample_data/hello.txt"}'
+make execute-ingestion input='{"file_path": "sample_data", "collection_name": "mydocs"}'
+```
+
+You can also trigger the workflow from the [Mistral Console](https://console.mistral.ai/build/workflows): select `document-ingestion`, click **Start Workflow**, and provide the input JSON.
+
+After ingestion, search directly (no workflow):
+
+```bash
+make search query="hello world"
+```
+
+## Folder layout
+
+```text
+examples/workflows/
+├── README.md              # This file
+├── worker.py              # Registers and runs all workflow examples
+├── start.py               # CLI to trigger a workflow execution
+└── search/                # Document ingestion workflow
+    ├── models.py
+    ├── activities.py
+    ├── workflow.py
+    └── README.md
+```
+
+## Design principles
+
+- **Workflows for ingestion** — document ingestion is long-running and benefits from durability, retries, and observability in the Mistral Console.
+- **Search stays direct** — search queries use `make search` / `entrypoints/search.py` for low latency. Workflows are not needed for queries.
+- **Small activity I/O** — the workflow passes file paths and collection names only. The full Search Toolkit pipeline runs inside a single `ingest_documents` activity; the activity returns a small result (`total_chunks`, `file_count`, …).
+- **Activities own all I/O** — filesystem access, API calls, and Vespa writes live in `activities.py`. The workflow body in `workflow.py` only orchestrates.
+
+## Designing ingestion workflows at scale
+
+Temporal stores every activity input and output in Postgres. **Do not split ingestion into one activity per pipeline stage** if each step passes document bytes, text, chunks, or embeddings — that duplicates large payloads in workflow history and does not scale.
+
+| Step | Cost | Where results should live |
+| --- | --- | --- |
+| Load from remote source | Medium | Object storage (S3) keyed by source URI + version |
+| Extract plain text, split | Cheap | Recompute inside the activity on retry — do not cache in Temporal |
+| OCR, embeddings | Expensive | External cache (Redis) keyed by `hash(file_bytes)`, not the raw content — implement outside this starter example |
+
+**Anti-pattern:** one `@workflows.activity` per pipeline stage with serialized `File` / `Document` dicts passed between them. That pattern only works for tiny demo files and fills Temporal storage quickly.
+
+### Per-stage retries
+
+This starter example runs the full pipeline inside a single activity so Temporal only stores small inputs and outputs (paths in, chunk counts out).
+
+If you need to **split the pipeline into separate activities** — for example, independent retry policies on OCR or embedding — you must still keep activity arguments and return values small. Pass references (file paths, S3 URIs, content hashes) between steps, and store intermediate or expensive-step results in external storage rather than in workflow history.
+
+## Adding a new workflow example
+
+1. Create a subdirectory under `examples/workflows/` (e.g. `my_example/`).
+2. Add `models.py`, `activities.py`, `workflow.py`, and a `README.md`.
+3. Export the workflow class from the package `__init__.py`.
+4. Register it in `EXAMPLE_WORKFLOWS` in `worker.py`.
+5. Add a Makefile target that calls `examples.workflows.start --workflow <name>`.
+
+## Troubleshooting
+
+| Issue | Fix |
+| --- | --- |
+| `Workflow not found` | Start `make start-examples` in a separate terminal, then retry |
+| `DEPLOYMENT_NAME is required` | Add `DEPLOYMENT_NAME=<your-project>` to `.env` |
+| Worker fails on startup | Ensure `imports_passed_through()` wraps activity imports in `workflow.py` when activities use `mistralai.client` |
+| Vespa errors during indexing | Run `make setup-vespa` before triggering ingestion |
+
+Enable verbose logging:
+
+```bash
+LOG_LEVEL=DEBUG make start-examples
+```
diff --git a/template/src/examples/workflows/__init__.py b/template/src/examples/workflows/__init__.py
new file mode 100644
index 0000000..1baee12
--- /dev/null
+++ b/template/src/examples/workflows/__init__.py
@@ -0,0 +1,5 @@
+"""Workflow-based examples for the search-starter-app.
+
+Each subdirectory is a self-contained workflow example. Run the worker with:
+    make start-examples
+"""
diff --git a/template/src/examples/workflows/search/README.md b/template/src/examples/workflows/search/README.md
new file mode 100644
index 0000000..26a17dd
--- /dev/null
+++ b/template/src/examples/workflows/search/README.md
@@ -0,0 +1,75 @@
+# Document Ingestion Workflow Example
+
+This example wraps the search-starter-app ingestion pipeline in a [Mistral Workflow](https://docs.mistral.ai/capabilities/workflows/), adding durability, observability, and retry support.
+
+It follows the recommended pattern for ingestion at scale: **small activity I/O**, with the full Search Toolkit pipeline running inside a single activity.
+
+## What It Demonstrates
+
+| Primitive | Where |
+| --- | --- |
+| `@workflows.workflow.define` | `workflow.py` — workflow registered in the Mistral Console |
+| `@workflows.workflow.entrypoint` | `workflow.py` — orchestration only; passes paths, not document content |
+| `@workflows.activity` | `activities.py` — all file I/O, embedding calls, and Vespa writes |
+| Pydantic input/output models | `models.py` — typed boundaries; workflow output is a small summary |
+| `workflows.run_worker()` | `worker.py` — registers the workflow with Temporal |
+
+## Architecture
+
+```
+IngestionWorkflow.run(IngestionInput)
+├── collect_document_paths activity   → list[str] paths only
+└── ingest_documents activity         → IngestionResult (counts + status)
+    └── Search Toolkit Pipeline:
+        load → extract → split → embed → index
+```
+
+The workflow never sees file bytes, extracted text, chunks, or embeddings — only paths and the final chunk count.
+
+## Prerequisites
+
+1. Install the workflows extra:
+   ```bash
+   make install-workflows
+   ```
+
+2. Vespa must be running:
+   ```bash
+   make setup-vespa
+   ```
+
+3. Set `MISTRAL_API_KEY` and `DEPLOYMENT_NAME` in your `.env` file.
+
+## Running the Example
+
+> The worker must be running before you trigger the workflow, otherwise the API returns `Workflow not found`.
+
+**Terminal 1 — start the worker (leave this running):**
+```bash
+make start-examples
+```
+
+**Terminal 2 — trigger ingestion:**
+```bash
+make execute-ingestion input='{"file_path": "sample_data/hello.txt"}'
+make execute-ingestion input='{"file_path": "sample_data"}'
+make execute-ingestion input='{"file_path": "sample_data/hello.txt", "collection_name": "mydocs"}'
+```
+
+You can also trigger the workflow from the [Mistral Console](https://console.mistral.ai/build/workflows) by selecting `document-ingestion`.
+
+## What not to do
+
+Do **not** split ingestion into one activity per pipeline stage while passing serialized documents between them. Temporal persists every activity input/output in Postgres; that pattern duplicates large payloads and does not scale. See the parent [workflows README](../README.md#designing-ingestion-workflows-at-scale).
+
+If you want to split the pipeline to enable retries for some stages, follow the [Handling Large Data](https://docs.mistral.ai/capabilities/workflows/guides/handling-large-data/) guide instead of passing document content between activities.
+
+## Key Difference from Direct Ingestion
+
+`make ingest` runs the same Search Toolkit pipeline synchronously. The workflow adds:
+
+- **Durability** — the worker can restart mid-ingestion and resume from the last completed activity
+- **Observability** — executions appear in the Mistral Console
+- **Retries** — the activity retries on transient errors without bloating Temporal with per-step document payloads
+
+Search queries remain direct (`make search`) — they do not need workflow orchestration.
diff --git a/template/src/examples/workflows/search/__init__.py b/template/src/examples/workflows/search/__init__.py
new file mode 100644
index 0000000..3d4c06d
--- /dev/null
+++ b/template/src/examples/workflows/search/__init__.py
@@ -0,0 +1,9 @@
+"""Workflows × Search integration example.
+
+Demonstrates wrapping the search-starter-app ingestion pipeline in a
+Mistral Workflow for durability, observability, and HITL support.
+"""
+
+from .workflow import IngestionWorkflow
+
+__all__ = ["IngestionWorkflow"]
diff --git a/template/src/examples/workflows/search/activities.py b/template/src/examples/workflows/search/activities.py
new file mode 100644
index 0000000..c4bd97f
--- /dev/null
+++ b/template/src/examples/workflows/search/activities.py
@@ -0,0 +1,172 @@
+"""Activities for the document ingestion workflow.
+
+All I/O lives here: file system access, embedding calls, and Vespa writes.
+The workflow body in workflow.py calls these as durable, retryable units.
+
+Stateless pipeline components and API clients are provided via Depends() so
+the SDK initialises them once at worker startup and reuses them across every
+activity execution. The only per-call objects are VespaSearchIndex and the
+Pipeline wrappers, because they are keyed to the collection_name input.
+"""
+
+import os
+from datetime import timedelta
+from functools import cache
+from pathlib import Path
+from typing import Any
+
+import mistralai.workflows as workflows
+from mistralai.client import Mistral
+from mistralai.search.toolkit.embedders import MistralEmbedder
+from mistralai.search.toolkit.ingestion.extractors import (
+    MistralOCRExtractor,
+    PlainTextExtractor,
+)
+from mistralai.search.toolkit.ingestion.loaders import FilesystemFileLoader
+from mistralai.search.toolkit.ingestion.pipelines import Pipeline
+from mistralai.search.toolkit.ingestion.text_splitters import (
+    MarkdownTextSplitter,
+    MarkdownTextSplitterConfig,
+)
+from mistralai.search.toolkit.plugins.vespa import VespaClientConfig
+from mistralai.workflows import Depends
+from vespa_app import app, vespa_endpoint
+
+from .models import IngestionResult
+
+_TEXT_SUFFIXES = {".txt", ".md", ".markdown", ".csv", ".json"}
+
+
+# ---------------------------------------------------------------------------
+# Dependency providers — each called once at worker startup
+# ---------------------------------------------------------------------------
+
+
+@cache
+def get_mistral_client() -> Mistral:
+    api_key = os.environ.get("MISTRAL_API_KEY", "")
+    if not api_key:
+        raise ValueError("MISTRAL_API_KEY is not set. Check your .env file.")
+    return Mistral(
+        api_key=api_key,
+        server_url=os.getenv("MISTRAL_API_URL", "https://api.mistral.ai"),
+    )
+
+
+def get_embedder() -> MistralEmbedder:
+    return MistralEmbedder(client=get_mistral_client())
+
+
+def get_ocr_extractor() -> MistralOCRExtractor:
+    return MistralOCRExtractor(client=get_mistral_client())
+
+
+def get_loader() -> FilesystemFileLoader:
+    return FilesystemFileLoader()
+
+
+def get_text_splitter() -> MarkdownTextSplitter:
+    return MarkdownTextSplitter(
+        MarkdownTextSplitterConfig(chunk_size=4096, chunk_overlap=50)
+    )
+
+
+# ---------------------------------------------------------------------------
+# Per-call pipeline factory
+# ---------------------------------------------------------------------------
+
+
+def _build_pipelines(
+    loader: FilesystemFileLoader,
+    text_splitter: MarkdownTextSplitter,
+    embedder: MistralEmbedder,
+    ocr_extractor: MistralOCRExtractor,
+    vector_store: Any,
+) -> tuple[Pipeline, Pipeline]:
+    """Assemble plain-text and OCR pipelines from shared components.
+
+    VespaSearchIndex (vector_store) is the only per-call argument because
+    it is keyed to the collection_name supplied at runtime.
+    """
+    shared = dict(
+        loader=loader,
+        text_splitter=text_splitter,
+        embedder=embedder,
+        stores=vector_store,
+    )
+    return (
+        Pipeline(extractor=PlainTextExtractor(), **shared),
+        Pipeline(extractor=ocr_extractor, **shared),
+    )
+
+
+# ---------------------------------------------------------------------------
+# Activities
+# ---------------------------------------------------------------------------
+
+
+@workflows.activity(
+    name="collect_document_paths",
+    start_to_close_timeout=timedelta(seconds=30),
+    retry_policy_max_attempts=1,
+)
+async def collect_document_paths(file_path: str) -> list[str]:
+    """Collect all file paths to ingest from a file or directory.
+
+    Returns string paths (Path objects are not serialisable across the
+    activity boundary) already split into two lists: text files and OCR files.
+    """
+    root = Path(file_path)
+    if root.is_file():
+        documents = [root]
+    elif root.is_dir():
+        documents = sorted(p for p in root.rglob("*") if p.is_file())
+    else:
+        raise ValueError(f"Path not found: {file_path}")
+
+    if not documents:
+        raise ValueError(f"No files found at: {file_path}")
+
+    return [str(p) for p in documents]
+
+
+@workflows.activity(
+    name="ingest_documents",
+    # Ingestion can be slow for large files or directories with many PDFs.
+    # 30 minutes is a generous upper bound; raise if processing very large corpora.
+    start_to_close_timeout=timedelta(minutes=30),
+    retry_policy_max_attempts=2,
+)
+async def ingest_documents(
+    paths: list[str],
+    collection_name: str,
+    embedder: MistralEmbedder = Depends(get_embedder),
+    ocr_extractor: MistralOCRExtractor = Depends(get_ocr_extractor),
+    loader: FilesystemFileLoader = Depends(get_loader),
+    text_splitter: MarkdownTextSplitter = Depends(get_text_splitter),
+) -> IngestionResult:
+    """Run the ingestion pipelines on a pre-collected list of file paths."""
+    vector_store = app.get_search_index(
+        VespaClientConfig(endpoint=vespa_endpoint()),
+        collection_name=collection_name,
+    )
+    plain_text_pipeline, ocr_pipeline = _build_pipelines(
+        loader, text_splitter, embedder, ocr_extractor, vector_store
+    )
+
+    documents = [Path(p) for p in paths]
+    text_docs = [p for p in documents if p.suffix.lower() in _TEXT_SUFFIXES]
+    ocr_docs = [p for p in documents if p.suffix.lower() not in _TEXT_SUFFIXES]
+
+    total_chunks = 0
+    if text_docs:
+        total_chunks += await plain_text_pipeline.run(documents=text_docs, use_checkpoint=False)
+    if ocr_docs:
+        total_chunks += await ocr_pipeline.run(documents=ocr_docs, use_checkpoint=False)
+
+    return IngestionResult(
+        status="success",
+        total_chunks=total_chunks,
+        file_count=len(documents),
+        collection_name=collection_name,
+    )
diff --git a/template/src/examples/workflows/search/models.py b/template/src/examples/workflows/search/models.py
new file mode 100644
index 0000000..ceb2e00
--- /dev/null
+++ b/template/src/examples/workflows/search/models.py
@@ -0,0 +1,22 @@
+"""Pydantic models for the document ingestion workflow."""
+
+from pydantic import BaseModel, Field
+
+
+class IngestionInput(BaseModel):
+    """Input accepted by the document-ingestion workflow."""
+
+    file_path: str = Field(description="Local path to a file or directory to ingest")
+    collection_name: str = Field(
+        default="exampledocs",
+        description="Vespa collection to index documents into",
+    )
+
+
+class IngestionResult(BaseModel):
+    """Output returned by the document-ingestion workflow."""
+
+    status: str = Field(description="'success' or 'error'")
+    total_chunks: int = Field(description="Total number of chunks indexed")
+    file_count: int = Field(description="Number of files processed")
+    collection_name: str = Field(description="Vespa collection that was used")
diff --git a/template/src/examples/workflows/search/workflow.py b/template/src/examples/workflows/search/workflow.py
new file mode 100644
index 0000000..c0c22f9
--- /dev/null
+++ b/template/src/examples/workflows/search/workflow.py
@@ -0,0 +1,36 @@
+"""Document ingestion workflow -- orchestration only, no I/O here."""
+
+import mistralai.workflows as workflows
+from mistralai.workflows import workflow
+
+# Activities import the search toolkit and mistralai.client (→ httpx), which
+# the Temporal sandbox blocks at workflow import time. imports_passed_through
+# lets the workflow reference activity stubs without loading I/O deps in-process.
+with workflow.unsafe.imports_passed_through():
+    from .activities import collect_document_paths, ingest_documents
+
+from .models import IngestionInput, IngestionResult  # noqa: E402
+
+
+@workflows.workflow.define(
+    name="document-ingestion",
+    workflow_display_name="Document Ingestion",
+    workflow_description=(
+        "Ingest documents from the local filesystem into the Vespa search index "
+        "using the Search Toolkit pipeline: load → extract → split → embed → index."
+    ),
+)
+class IngestionWorkflow:
+    """Workflow that wraps the search-starter-app ingestion pipeline.
+
+    All I/O (file loading, LLM calls, Vespa writes) lives in activities.py so
+    the workflow body stays deterministic and replayable by the Temporal engine.
+    """
+
+    @workflows.workflow.entrypoint
+    async def run(self, params: IngestionInput) -> IngestionResult:
+        paths = await collect_document_paths(file_path=params.file_path)
+        return await ingest_documents(
+            paths=paths,
+            collection_name=params.collection_name,
+        )
diff --git a/template/src/examples/workflows/start.py b/template/src/examples/workflows/start.py
new file mode 100644
index 0000000..7397fc4
--- /dev/null
+++ b/template/src/examples/workflows/start.py
@@ -0,0 +1,92 @@
+"""Trigger a workflow execution from the command line.
+
+Requires the optional workflows extra:
+    uv sync --extra workflows
+"""
+# ruff: noqa: E402
+
+import argparse
+import asyncio
+import json
+import os
+
+from dotenv import load_dotenv
+
+load_dotenv(override=True)
+
+from mistralai.extra.workflows import WorkflowEncodingConfig, configure_workflow_encoding
+from mistralai.workflows.client import get_mistral_client
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Trigger a workflow execution.")
+    parser.add_argument(
+        "--workflow",
+        default="document-ingestion",
+        help="Workflow name (default: document-ingestion)",
+    )
+    parser.add_argument(
+        "--input",
+        default=r"{}",
+        help=r'Input data as a JSON string (e.g. \'{"file_path": "sample_data/hello.txt"}\')',
+    )
+    return parser.parse_args()
+
+
+async def main() -> None:
+    args = parse_args()
+
+    try:
+        raw_input = json.loads(args.input)
+    except json.JSONDecodeError as exc:
+        raise SystemExit(
+            f"Error: invalid JSON for --input: {exc.args[0]}\n"
+            f"  Received: {args.input!r}\n"
+            f'  Example:  --input \'{{"file_path": "sample_data/hello.txt"}}\''
+        ) from exc
+
+    raw_input = raw_input or {}
+    if not isinstance(raw_input, dict):
+        raise SystemExit(
+            f"Error: --input must be a JSON object, got {type(raw_input).__name__}"
+        )
+
+    api_key = os.environ.get("MISTRAL_API_KEY", "")
+    if not api_key:
+        raise SystemExit("Error: MISTRAL_API_KEY is not set. Check your .env file.")
+
+    deployment_name = os.environ.get("DEPLOYMENT_NAME")
+    if not deployment_name:
+        raise SystemExit(
+            "Error: DEPLOYMENT_NAME is not set. Add it to your .env file, e.g.:\n"
+            "  DEPLOYMENT_NAME=my-search-project"
+        )
+
+    client = get_mistral_client(
+        api_key=api_key,
+        server_url=os.environ.get("MISTRAL_API_URL", "https://api.mistral.ai"),
+    )
+
+    await configure_workflow_encoding(WorkflowEncodingConfig(), client=client)
+
+    try:
+        result = await client.workflows.execute_workflow_and_wait_async(
+            workflow_identifier=args.workflow,
+            input=raw_input,
+            deployment_name=deployment_name,
+        )
+    except Exception as exc:
+        if "Workflow not found" in str(exc) or "404" in str(exc):
+            raise SystemExit(
+                f"Error: workflow '{args.workflow}' not found.\n"
+                "Start the examples worker first (separate terminal):\n"
+                "  make start-examples\n"
+                "Then retry this command."
+            ) from exc
+        raise
+
+    print(f"Result: {result}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/template/src/examples/workflows/worker.py b/template/src/examples/workflows/worker.py
new file mode 100644
index 0000000..60c2461
--- /dev/null
+++ b/template/src/examples/workflows/worker.py
@@ -0,0 +1,36 @@
+"""Start a worker that registers all example workflows for search-starter-app."""
+# ruff: noqa: E402
+
+import asyncio
+import os
+import sys
+
+from dotenv import load_dotenv
+
+load_dotenv(override=True)
+
+import mistralai.workflows as workflows
+
+from examples.workflows.search import IngestionWorkflow
+
+EXAMPLE_WORKFLOWS = [IngestionWorkflow]
+
+
+async def main() -> None:
+    if not os.environ.get("DEPLOYMENT_NAME"):
+        print(
+            "Error: DEPLOYMENT_NAME is not set. Add it to your .env file, e.g.:\n"
+            "  DEPLOYMENT_NAME=my-search-project",
+            file=sys.stderr,
+        )
+        raise SystemExit(1)
+
+    names = [wf.__name__ for wf in EXAMPLE_WORKFLOWS]
+    print(
+        f"Starting worker with {len(EXAMPLE_WORKFLOWS)} example(s): {', '.join(names)}"
+    )
+    await workflows.run_worker(EXAMPLE_WORKFLOWS)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())