updates, handle async activities definitions

Signed-off-by: Filinto Duran <[email protected]>

Author: filintod

examples/workflow-async/README.md

@@ -1,7 +1,7 @@
 # Dapr Workflow Async Examples (Python)
 
 These examples mirror `examples/workflow/` but author orchestrators with `async def` using the
-async workflow APIs. Activities remain regular functions unless noted.
+async workflow APIs. Activities can be either sync or async functions.
 
 ## Prerequisites
 
@@ -34,6 +34,7 @@ How to run:
   - `dapr run --app-id wf_async_gather -- /Users/filinto/diagrid/python-sdk/examples/workflow-async/.venv/bin/python fan_out_fan_in_with_gather.py`
   - `dapr run --app-id wf_async_approval -- /Users/filinto/diagrid/python-sdk/examples/workflow-async/.venv/bin/python human_approval.py`
   - `dapr run --app-id wf_ctx_interceptors -- /Users/filinto/diagrid/python-sdk/examples/workflow-async/.venv/bin/python context_interceptors_example.py`
+  - `dapr run --app-id wf_async_http -- /Users/filinto/diagrid/python-sdk/examples/workflow-async/.venv/bin/python async_http_activity.py`
 
 ## Examples
 
@@ -44,9 +45,42 @@ How to run:
 - **fan_out_fan_in_with_gather.py**: Parallel execution using `ctx.when_all()` 
 - **human_approval.py**: Workflow waiting for external event to proceed
 - **context_interceptors_example.py**: Context propagation using interceptors (tenant, request ID, etc.)
+- **async_http_activity.py**: Async activities performing I/O-bound operations (HTTP requests with aiohttp)
 
 Notes:
-- Orchestrators use `await ctx.activity(...)`, `await ctx.sleep(...)`, `await ctx.when_all/when_any(...)`, etc.
+- Orchestrators use `await ctx.activity(...)`, `await ctx.create_timer(...)`, `await ctx.when_all/when_any(...)`, etc.
 - No event loop is started manually; the Durable Task worker drives the async orchestrators.
 - You can also launch instances using `DaprWorkflowClient` as in the non-async examples.
 - The interceptors example demonstrates how to propagate context (tenant, request ID) across workflow and activity boundaries using the wrapper pattern to avoid contextvar loss.
+
+## Async Activities
+
+Activities can be either synchronous or asynchronous functions. Async activities are useful for I/O-bound operations like HTTP requests, database queries, or file operations:
+
+```python
+from dapr.ext.workflow import WorkflowActivityContext
+
+# Synchronous activity
+@wfr.activity
+def sync_activity(ctx: WorkflowActivityContext, data: str) -> str:
+    return data.upper()
+
+# Asynchronous activity
+@wfr.activity
+async def async_activity(ctx: WorkflowActivityContext, data: str) -> str:
+    # Perform async I/O operations
+    async with aiohttp.ClientSession() as session:
+        async with session.get(f"https://api.example.com/{data}") as response:
+            result = await response.json()
+    return result
+```
+
+Both sync and async activities are registered the same way using the `@wfr.activity` decorator. Orchestrators call them identically regardless of whether they're sync or async - the SDK handles the execution automatically.
+
+**When to use async activities:**
+- HTTP requests or API calls
+- Database queries
+- File I/O operations
+- Any I/O-bound work that benefits from async/await
+
+See `async_http_activity.py` for a complete example.

examples/workflow-async/async_http_activity.py

@@ -0,0 +1,153 @@
+# -*- coding: utf-8 -*-
+"""
+Copyright 2025 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.
+"""
+
+from dapr.ext.workflow import (  # noqa: E402
+    AsyncWorkflowContext,
+    DaprWorkflowClient,
+    WorkflowActivityContext,
+    WorkflowRuntime,
+    WorkflowStatus,
+)
+
+"""Example demonstrating async activities with HTTP requests.
+
+This example shows how to use async activities to perform I/O-bound operations
+like HTTP requests without blocking the worker thread pool.
+"""
+
+
+wfr = WorkflowRuntime()
+
+
+@wfr.activity(name='fetch_url')
+async def fetch_url(ctx: WorkflowActivityContext, url: str) -> dict:
+    """Async activity that fetches data from a URL.
+
+    This demonstrates using aiohttp for non-blocking HTTP requests.
+    In production, you would handle errors, timeouts, and retries.
+    """
+    try:
+        import aiohttp
+    except ImportError:
+        # Fallback if aiohttp is not installed
+        return {
+            'url': url,
+            'status': 'error',
+            'message': 'aiohttp not installed. Install with: pip install aiohttp',
+        }
+
+    try:
+        async with aiohttp.ClientSession() as session:
+            async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as response:
+                status = response.status
+                if status == 200:
+                    # For JSON responses
+                    try:
+                        data = await response.json()
+                        return {'url': url, 'status': status, 'data': data}
+                    except Exception:
+                        # For text responses
+                        text = await response.text()
+                        return {
+                            'url': url,
+                            'status': status,
+                            'length': len(text),
+                            'preview': text[:100],
+                        }
+                else:
+                    return {'url': url, 'status': status, 'error': 'HTTP error'}
+    except Exception as e:
+        return {'url': url, 'status': 'error', 'message': str(e)}
+
+
+@wfr.activity(name='process_data')
+def process_data(ctx: WorkflowActivityContext, data: dict) -> dict:
+    """Sync activity that processes fetched data.
+
+    This shows that sync and async activities can coexist in the same workflow.
+    """
+    return {
+        'processed': True,
+        'url_count': len([k for k in data if k.startswith('url_')]),
+        'summary': f'Processed {len(data)} items',
+    }
+
+
+@wfr.async_workflow(name='fetch_multiple_urls_async')
+async def fetch_multiple_urls(ctx: AsyncWorkflowContext, urls: list[str]) -> dict:
+    """Orchestrator that fetches multiple URLs in parallel using async activities.
+
+    This demonstrates:
+    - Calling async activities from async workflows
+    - Fan-out/fan-in pattern with async activities
+    - Mixing async and sync activities
+    """
+    # Fan-out: Schedule all URL fetches in parallel
+    fetch_tasks = [ctx.call_activity(fetch_url, input=url) for url in urls]
+
+    # Fan-in: Wait for all to complete
+    results = await ctx.when_all(fetch_tasks)
+
+    # Create a dictionary of results
+    data = {f'url_{i}': result for i, result in enumerate(results)}
+
+    # Process the aggregated data with a sync activity
+    summary = await ctx.call_activity(process_data, input=data)
+
+    return {'results': data, 'summary': summary}
+
+
+def main():
+    """Run the example workflow."""
+    # Example URLs to fetch (using httpbin.org for testing)
+    test_urls = [
+        'https://httpbin.org/json',
+        'https://httpbin.org/uuid',
+        'https://httpbin.org/user-agent',
+    ]
+
+    wfr.start()
+    client = DaprWorkflowClient()
+
+    try:
+        instance_id = 'async_http_activity_example'
+        print(f'Starting workflow {instance_id}...')
+
+        # Schedule the workflow
+        client.schedule_new_workflow(
+            workflow=fetch_multiple_urls, instance_id=instance_id, input=test_urls
+        )
+
+        # Wait for completion
+        wf_state = client.wait_for_workflow_completion(instance_id, timeout_in_seconds=60)
+
+        print(f'\nWorkflow status: {wf_state.runtime_status}')
+
+        if wf_state.runtime_status == WorkflowStatus.COMPLETED:
+            print(f'Workflow output: {wf_state.serialized_output}')
+            print('\n✓ Workflow completed successfully!')
+        else:
+            print('✗ Workflow did not complete successfully')
+            return 1
+
+    finally:
+        wfr.shutdown()
+
+    return 0
+
+
+if __name__ == '__main__':
+    import sys
+
+    sys.exit(main())

examples/workflow-async/context_interceptors_example.py

@@ -52,6 +52,7 @@ def wrapper():
 from __future__ import annotations
 
 import contextvars
+from dataclasses import replace
 from typing import Any, Callable
 
 from dapr.ext.workflow import (
@@ -102,15 +103,7 @@ def schedule_new_workflow(
         print('[Client] Scheduling workflow with metadata:', metadata)
 
         # Set metadata on the request (runtime will wrap in envelope)
-        modified_request = ScheduleWorkflowRequest(
-            workflow_name=request.workflow_name,
-            input=request.input,
-            instance_id=request.instance_id,
-            start_at=request.start_at,
-            reuse_id_policy=request.reuse_id_policy,
-            metadata=metadata,
-        )
-        return nxt(modified_request)
+        return nxt(replace(request, metadata=metadata))
 
 
 class ContextWorkflowOutboundInterceptor(BaseWorkflowOutboundInterceptor):
@@ -126,15 +119,7 @@ def call_activity(
         ctx = get_request_context()
         metadata = ctx.copy() if ctx else {}
 
-        return nxt(
-            CallActivityRequest(
-                activity_name=request.activity_name,
-                input=request.input,
-                retry_policy=request.retry_policy,
-                workflow_ctx=request.workflow_ctx,
-                metadata=metadata,
-            )
-        )
+        return nxt(replace(request, metadata=metadata))
 
     def call_child_workflow(
         self, request: CallChildWorkflowRequest, nxt: Callable[[CallChildWorkflowRequest], Any]
@@ -143,15 +128,7 @@ def call_child_workflow(
         ctx = get_request_context()
         metadata = ctx.copy() if ctx else {}
 
-        return nxt(
-            CallChildWorkflowRequest(
-                workflow_name=request.workflow_name,
-                input=request.input,
-                instance_id=request.instance_id,
-                workflow_ctx=request.workflow_ctx,
-                metadata=metadata,
-            )
-        )
+        return nxt(replace(request, metadata=metadata))
 
 
 class ContextRuntimeInterceptor(BaseRuntimeInterceptor):

examples/workflow/context_interceptors_example.py

@@ -16,8 +16,8 @@ def wrapper():
             setup_context()
             try:
                 gen = nxt(request)
-                result = yield from gen  # Keep context alive
-                return result  # MUST return to propagate workflow output
+                result = yield from gen     # Keep context alive
+                return result               # MUST return to propagate workflow output
             finally:
                 cleanup_context()
         return wrapper()
@@ -52,6 +52,7 @@ def wrapper():
 
 import contextvars
 import json
+from dataclasses import replace
 from typing import Any, Callable
 
 from dapr.ext.workflow import (
@@ -100,15 +101,7 @@ def schedule_new_workflow(
         print('[Client] Scheduling workflow with metadata:', metadata)
 
         # Set metadata on the request (runtime will wrap in envelope)
-        modified_request = ScheduleWorkflowRequest(
-            workflow_name=request.workflow_name,
-            input=request.input,  # Input unchanged - metadata stored separately
-            instance_id=request.instance_id,
-            start_at=request.start_at,
-            reuse_id_policy=request.reuse_id_policy,
-            metadata=metadata,
-        )
-        return nxt(modified_request)
+        return nxt(replace(request, metadata=metadata))
 
 
 class ContextWorkflowOutboundInterceptor(BaseWorkflowOutboundInterceptor):
@@ -126,15 +119,8 @@ def call_child_workflow(
 
         print('[Outbound] Calling child workflow with metadata:', metadata)
 
-        return nxt(
-            CallChildWorkflowRequest(
-                workflow_name=request.workflow_name,
-                input=request.input,  # Input unchanged - metadata stored separately
-                instance_id=request.instance_id,
-                workflow_ctx=request.workflow_ctx,
-                metadata=metadata,
-            )
-        )
+        # Use dataclasses.replace() to create a modified copy
+        return nxt(replace(request, metadata=metadata))
 
     def call_activity(
         self, request: CallActivityRequest, nxt: Callable[[CallActivityRequest], Any]
@@ -147,15 +133,8 @@ def call_activity(
         print(f'  -- input: {request.input}')
         print(f'  -- metadata: {metadata}')
 
-        return nxt(
-            CallActivityRequest(
-                activity_name=request.activity_name,
-                input=request.input,  # Input unchanged - metadata stored separately
-                retry_policy=request.retry_policy,
-                workflow_ctx=request.workflow_ctx,
-                metadata=metadata,
-            )
-        )
+        # Use dataclasses.replace() to create a modified copy
+        return nxt(replace(request, metadata=metadata))
 
 
 class ContextRuntimeInterceptor(BaseRuntimeInterceptor):