Add support for async workflow activities

.gitignore

@@ -142,3 +142,6 @@ tests/integration/.binding-data/
 
 # Crypto test material generated at test time (see tests/crypto_utils.py)
 tests/integration/keys/
+
+# Generated benchmark report (machine-specific, regenerated by bench_async_activities.py)
+ext/dapr-ext-workflow/benchmarks/RESULTS.md

examples/workflow/async_activities.py

@@ -0,0 +1,104 @@
+# -*- coding: utf-8 -*-
+# Copyright 2026 The Dapr Authors
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#     http://www.apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Async activities running alongside sync ones in a single workflow.
+
+Starts three async activities that do an HTTP request, then a sync activity that
+sums up the results. Shows that sync and async activities work side by side.
+
+Run with:
+
+    dapr run --app-id async-activities --app-protocol grpc --dapr-grpc-port 50001 \\
+        -- python async_activities.py
+"""
+
+from __future__ import annotations
+
+from time import sleep
+
+import dapr.ext.workflow as wf
+import httpx
+from pydantic import BaseModel
+
+wfr = wf.WorkflowRuntime()
+
+
+class FetchRequest(BaseModel):
+    url: str
+    timeout_seconds: float = 5.0
+
+
+class FetchResult(BaseModel):
+    url: str
+    status_code: int
+    body_length: int
+
+
+@wfr.workflow(name='parallel_fetch_workflow')
+def parallel_fetch_workflow(ctx: wf.DaprWorkflowContext, urls: list[str]):
+    fetch_tasks = [
+        ctx.call_activity(fetch_url, input=FetchRequest(url=url).model_dump()) for url in urls
+    ]
+    results = yield wf.when_all(fetch_tasks)
+    summary = yield ctx.call_activity(summarize_fetches, input=results)
+    return summary
+
+
+@wfr.activity(name='fetch_url')
+async def fetch_url(ctx: wf.WorkflowActivityContext, request: FetchRequest) -> dict:
+    """Async activity: fetch a URL with httpx. Multiple instances run concurrently."""
+    async with httpx.AsyncClient(timeout=request.timeout_seconds) as client:
+        response = await client.get(request.url)
+    result = FetchResult(
+        url=request.url,
+        status_code=response.status_code,
+        body_length=len(response.content),
+    )
+    print(
+        f'[async] fetched {result.url} -> {result.status_code} ({result.body_length}B)', flush=True
+    )
+    return result.model_dump()
+
+
+@wfr.activity(name='summarize_fetches')
+def summarize_fetches(ctx: wf.WorkflowActivityContext, results: list[dict]) -> str:
+    """Sync activity: runs in the sync-fallback thread pool. Unchanged from before."""
+    total_bytes = sum(r['body_length'] for r in results)
+    summary = f'fetched {len(results)} URLs, total {total_bytes} bytes'
+    print(f'[sync] {summary}', flush=True)
+    return summary
+
+
+def main() -> None:
+    urls = [
+        'https://example.com',
+        'https://example.org',
+        'https://example.net',
+    ]
+
+    wfr.start()
+    sleep(5)  # wait for workflow runtime to start
+
+    wf_client = wf.DaprWorkflowClient()
+    instance_id = wf_client.schedule_new_workflow(workflow=parallel_fetch_workflow, input=urls)
+    print(f'Workflow started. Instance ID: {instance_id}')
+
+    state = wf_client.wait_for_workflow_completion(instance_id, timeout_in_seconds=60)
+    assert state is not None
+    print(f'Workflow completed! Status: {state.runtime_status.name}')
+    print(f'Workflow result: {state.serialized_output.strip(chr(34))}')
+
+    wfr.shutdown()
+
+
+if __name__ == '__main__':
+    main()

ext/dapr-ext-strands/pyproject.toml

@@ -21,6 +21,7 @@ classifiers = [
 dependencies = [
     "dapr",
     "strands-agents>=1.30.0,<2.0.0",
+    "strands-agents-tools>=0.2.22,<1.0.0",
     "python-ulid>=3.0.0,<4.0.0",
     "msgpack-python>=0.4.5,<1.0.0",
 ]

ext/dapr-ext-workflow/AGENTS.md

@@ -105,6 +105,24 @@ The entry point for registration and lifecycle:
 
 Internally wraps user functions: workflow functions get a `DaprWorkflowContext`, activity functions get a `WorkflowActivityContext`. Tracks registration state via `_workflow_registered` / `_activity_registered` attributes on functions to prevent double registration.
 
+#### Sync and async activities
+
+Activities can be either `def my_activity(ctx, inp)` or `async def my_activity(ctx, inp)`. At registration, `_make_activity_wrapper` calls `_is_async_callable(fn)` to detect async-ness. That helper unwraps `functools.partial`, `@functools.wraps` chains, and callable-class `__call__` so common decorator patterns route correctly. The wrapper is built `async def` or `def` to match, then stored in the registry.
+
+At dispatch time (the gRPC stream loop in `_durabletask/worker.py`), `is_async_callable(activity_fn)` on the wrapper selects between two handlers.
+
+- **Async activities** go through `_execute_activity_async`, then `_ActivityExecutor.execute_async`, which awaits `fn(...)` directly on the event loop. The gRPC response is delivered via `loop.run_in_executor(self._async_worker_manager.thread_pool, stub.CompleteActivityTask, ...)` — the same pool sync activities use, sized by `maximum_thread_pool_workers`.
+- **Sync activities** go through `_execute_activity`, dispatched to the thread pool by `_AsyncWorkerManager._run_func`. The activity runs on a worker thread, and the response is delivered from the same thread.
+
+Workflow (orchestrator) functions must remain generators (`def` with `yield`). They cannot be `async def` because durabletask's deterministic replay depends on synchronous generator semantics. Only activities support async.
+
+**Decorator ordering gotcha.** Stacking `@wfr.activity` over `@alternate_name(...)` over `async def` works because `@alternate_name` now emits an `async def innerfn` when the wrapped function is async. A user-written decorator that wraps an async function in a sync `def` (without `@functools.wraps` exposing `__wrapped__`) defeats `_is_async_callable`, routes the activity to the sync path, and produces an un-awaited coroutine. Such decorators should use `@functools.wraps(fn)` so the unwrap walks through them.
+
+**`maximum_thread_pool_workers` covers both paths.** This knob sizes the worker thread pool used for sync-activity bodies and for async-activity gRPC response sends. Mixed workloads with long-running sync activities can starve async response delivery (and vice versa) since they share the pool — size to the sum of peak sync activity concurrency and peak in-flight async response sends.
+
+**Concurrency sizing and load characterization.** See `docs/concurrency.md` for sizing recommendations (`maximum_concurrent_activity_work_items`, `maximum_thread_pool_workers`) and an async-vs-sync decision tree. The `benchmarks/` directory ships `bench_async_activities.py`; re-run it locally before claiming a perf regression. The generated `RESULTS.md` is gitignored because numbers are machine-specific; see `docs/concurrency.md` for the regen command.
+
+
 ### DaprWorkflowClient (`dapr_workflow_client.py`)
 
 Client for workflow lifecycle management:
@@ -163,7 +181,7 @@ Retry configuration for activities and child workflows:
 1. **Registration**: User decorates functions with `@wfr.workflow` / `@wfr.activity`. The runtime wraps them and stores them in the durabletask worker's registry.
 2. **Startup**: `wfr.start()` opens a gRPC stream to the Dapr sidecar. The worker polls for work items.
 3. **Scheduling**: Client calls `schedule_new_workflow(fn, input=...)`. The function's name (or `_dapr_alternate_name`) is sent to the backend.
-4. **Execution**: The durabletask engine dispatches work items. Workflow functions are Python **generators** that `yield` tasks (activity calls, timers, child workflows). The engine records history; on replay, yielded tasks return cached results without re-executing.
+4. **Execution**: The durabletask engine dispatches work items. Workflow functions are Python **generators** that `yield` tasks (activity calls, timers, child workflows). Activity functions are either sync (dispatched to the worker's thread pool) or `async def` (awaited directly on the worker's event loop). The engine records history; on replay, yielded tasks return cached results without re-executing.
 5. **Determinism**: Workflows must be deterministic — no random, no wall-clock time, no I/O. Use `ctx.current_utc_datetime` instead of `datetime.now()`. Use `ctx.is_replaying` to guard side effects like logging.
 6. **Completion**: Client polls via `wait_for_workflow_completion()` or `get_workflow_state()`.
 
@@ -191,6 +209,7 @@ Two example directories exercise workflows:
   - `cross-app1.py`, `cross-app2.py`, `cross-app3.py` — cross-app calls
   - `versioning.py` — workflow versioning with `is_patched()`
   - `simple_aio_client.py` — async client variant
+  - `async_activities.py` — `async def` activities (HTTP fan-out with `httpx.AsyncClient`)
 
 ## Testing