allow async activity definitions

Signed-off-by: Filinto Duran <1373693+filintod@users.noreply.github.com>

Author: filintod

README.md

@@ -138,6 +138,35 @@ Orchestrations are implemented using ordinary Python functions that take an `Orc
 
 Activities are implemented using ordinary Python functions that take an `ActivityContext` as their first parameter. Activity functions are scheduled by orchestrations and have at-least-once execution guarantees, meaning that they will be executed at least once but may be executed multiple times in the event of a transient failure. Activity functions are where the real "work" of any orchestration is done.
 
+#### 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 durabletask.task import ActivityContext
+
+# Synchronous activity
+def sync_activity(ctx: ActivityContext, data: str) -> str:
+    return data.upper()
+
+# Asynchronous activity
+async def async_activity(ctx: ActivityContext, 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:
+
+```python
+worker.add_activity(sync_activity)
+worker.add_activity(async_activity)
+```
+
+Orchestrators call them identically regardless of whether they're sync or async - the SDK handles the execution automatically.
+
 ### Durable timers
 
 Orchestrations can schedule durable timers using the `create_timer` API. These timers are durable, meaning that they will survive orchestrator restarts and will fire even if the orchestrator is not actively in memory. Durable timers can be of any duration, from milliseconds to months.

durabletask/aio/ASYNCIO_INTERNALS.md

@@ -292,7 +292,7 @@ Adding sandbox coverage:
 
 ## Interop Checklist (Async ↔ Generator)
 
-- Activities: identical behavior; only authoring differs (`yield` vs `await`).
+- Activities: identical behavior; only authoring differs (`yield` vs `await`). Activities themselves can be either sync or async functions and work identically from both generator and async orchestrators.
 - Timers: map to the same `createTimer` actions.
 - External events: same semantics for buffering and completion.
 - Sub‑orchestrators: same create/complete/fail events.

durabletask/task.py

@@ -7,7 +7,7 @@
 import math
 from abc import ABC, abstractmethod
 from datetime import datetime, timedelta
-from typing import Any, Callable, Generator, Generic, Optional, TypeVar, Union
+from typing import Any, Awaitable, Callable, Generator, Generic, Optional, TypeVar, Union
 
 import durabletask.internal.helpers as pbh
 import durabletask.internal.orchestrator_service_pb2 as pb
@@ -499,7 +499,10 @@ def task_id(self) -> int:
 Orchestrator = Callable[[OrchestrationContext, TInput], Union[Generator[Task, Any, Any], TOutput]]
 
 # Activities are simple functions that can be scheduled by orchestrators
-Activity = Callable[[ActivityContext, TInput], TOutput]
+Activity = Union[
+    Callable[[ActivityContext, TInput], TOutput],
+    Callable[[ActivityContext, TInput], Awaitable[TOutput]],
+]
 
 
 class RetryPolicy:

durabletask/worker.py

@@ -692,7 +692,7 @@ def _execute_orchestrator(
                 f"Failed to deliver orchestrator response for '{req.instanceId}' to sidecar: {ex}"
             )
 
-    def _execute_activity(
+    async def _execute_activity(
         self,
         req: pb.ActivityRequest,
         stub: stubs.TaskHubSidecarServiceStub,
@@ -701,7 +701,7 @@ def _execute_activity(
         instance_id = req.orchestrationInstance.instanceId
         try:
             executor = _ActivityExecutor(self._registry, self._logger)
-            result = executor.execute(instance_id, req.name, req.taskId, req.input.value)
+            result = await executor.execute(instance_id, req.name, req.taskId, req.input.value)
             res = pb.ActivityResponse(
                 instanceId=instance_id,
                 taskId=req.taskId,
@@ -1441,7 +1441,7 @@ def __init__(self, registry: _Registry, logger: logging.Logger):
         self._registry = registry
         self._logger = logger
 
-    def execute(
+    async def execute(
         self,
         orchestration_id: str,
         name: str,
@@ -1459,8 +1459,11 @@ def execute(
         activity_input = shared.from_json(encoded_input) if encoded_input else None
         ctx = task.ActivityContext(orchestration_id, task_id)
 
-        # Execute the activity function
-        activity_output = fn(ctx, activity_input)
+        # Execute the activity function (sync or async)
+        if inspect.iscoroutinefunction(fn):
+            activity_output = await fn(ctx, activity_input)
+        else:
+            activity_output = fn(ctx, activity_input)
 
         encoded_output = shared.to_json(activity_output) if activity_output is not None else None
         chars = len(encoded_output) if encoded_output else 0

tests/aio/test_context_simple.py

@@ -15,6 +15,7 @@
 These tests focus on the actual implementation rather than expected features.
 """
 
+import asyncio
 import random
 import uuid
 from datetime import datetime, timedelta
@@ -308,3 +309,64 @@ def test_context_repr(self):
         repr_str = repr(self.ctx)
         assert "AsyncWorkflowContext" in repr_str
         assert "test-instance-123" in repr_str
+
+
+class TestAsyncActivities:
+    """Test async activities called from AsyncWorkflowContext."""
+
+    def setup_method(self):
+        """Set up test fixtures."""
+        self.mock_base_ctx = Mock(spec=dt_task.OrchestrationContext)
+        self.mock_base_ctx.instance_id = "test-instance-123"
+        self.mock_base_ctx.current_utc_datetime = datetime(2023, 1, 1, 12, 0, 0)
+        self.mock_base_ctx.is_replaying = False
+        self.mock_base_ctx.is_suspended = False
+
+    def test_async_activity_call(self):
+        """Test AsyncWorkflowContext calling async activity"""
+
+        async def async_activity(ctx: dt_task.ActivityContext, input_data: str):
+            await asyncio.sleep(0.001)
+            return input_data.upper()
+
+        ctx = AsyncWorkflowContext(self.mock_base_ctx)
+        awaitable = ctx.call_activity(async_activity, input="test")
+
+        assert isinstance(awaitable, ActivityAwaitable)
+        assert awaitable._activity_fn == async_activity
+        assert awaitable._input == "test"
+
+    def test_async_activity_with_when_all(self):
+        """Test when_all with async activities"""
+
+        async def async_activity(ctx: dt_task.ActivityContext, input_data: int):
+            await asyncio.sleep(0.001)
+            return input_data * 2
+
+        ctx = AsyncWorkflowContext(self.mock_base_ctx)
+
+        # Create multiple async activity awaitables
+        awaitables = [ctx.call_activity(async_activity, input=i) for i in range(3)]
+        when_all = ctx.when_all(awaitables)
+
+        assert isinstance(when_all, WhenAllAwaitable)
+
+    def test_async_activity_with_when_any(self):
+        """Test when_any with async activities"""
+
+        async def async_activity_fast(ctx: dt_task.ActivityContext, _):
+            await asyncio.sleep(0.001)
+            return "fast"
+
+        async def async_activity_slow(ctx: dt_task.ActivityContext, _):
+            await asyncio.sleep(0.1)
+            return "slow"
+
+        ctx = AsyncWorkflowContext(self.mock_base_ctx)
+
+        # Create async activity awaitables
+        fast_awaitable = ctx.call_activity(async_activity_fast, input=None)
+        slow_awaitable = ctx.call_activity(async_activity_slow, input=None)
+        when_any = ctx.when_any([fast_awaitable, slow_awaitable])
+
+        assert isinstance(when_any, WhenAnyAwaitable)

tests/durabletask/test_activity_executor.py

@@ -1,6 +1,7 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
+import asyncio
 import json
 import logging
 from typing import Any, Optional, Tuple
@@ -26,7 +27,9 @@ def test_activity(ctx: task.ActivityContext, test_input: Any):
 
     activity_input = "Hello, 世界!"
     executor, name = _get_activity_executor(test_activity)
-    result = executor.execute(TEST_INSTANCE_ID, name, TEST_TASK_ID, json.dumps(activity_input))
+    result = asyncio.run(
+        executor.execute(TEST_INSTANCE_ID, name, TEST_TASK_ID, json.dumps(activity_input))
+    )
     assert result is not None
 
     result_input, result_orchestration_id, result_task_id = json.loads(result)
@@ -43,7 +46,7 @@ def test_activity(ctx: task.ActivityContext, _):
 
     caught_exception: Optional[Exception] = None
     try:
-        executor.execute(TEST_INSTANCE_ID, "Bogus", TEST_TASK_ID, None)
+        asyncio.run(executor.execute(TEST_INSTANCE_ID, "Bogus", TEST_TASK_ID, None))
     except Exception as ex:
         caught_exception = ex
 
@@ -56,3 +59,97 @@ def _get_activity_executor(fn: task.Activity) -> Tuple[worker._ActivityExecutor,
     name = registry.add_activity(fn)
     executor = worker._ActivityExecutor(registry, TEST_LOGGER)
     return executor, name
+
+
+def test_async_activity_basic():
+    """Validates basic async activity execution"""
+
+    async def async_activity(ctx: task.ActivityContext, test_input: str):
+        # Simple async activity that returns modified input
+        return f"async:{test_input}"
+
+    activity_input = "test"
+    executor, name = _get_activity_executor(async_activity)
+
+    # Run the async executor
+    result = asyncio.run(
+        executor.execute(TEST_INSTANCE_ID, name, TEST_TASK_ID, json.dumps(activity_input))
+    )
+    assert result is not None
+
+    result_output = json.loads(result)
+    assert result_output == "async:test"
+
+
+def test_async_activity_with_input():
+    """Validates async activity with complex input/output"""
+
+    async def async_activity(ctx: task.ActivityContext, test_input: dict):
+        # Return all activity inputs back as the output
+        return {
+            "input": test_input,
+            "orchestration_id": ctx.orchestration_id,
+            "task_id": ctx.task_id,
+            "processed": True,
+        }
+
+    activity_input = {"key": "value", "number": 42}
+    executor, name = _get_activity_executor(async_activity)
+    result = asyncio.run(
+        executor.execute(TEST_INSTANCE_ID, name, TEST_TASK_ID, json.dumps(activity_input))
+    )
+    assert result is not None
+
+    result_data = json.loads(result)
+    assert result_data["input"] == activity_input
+    assert result_data["orchestration_id"] == TEST_INSTANCE_ID
+    assert result_data["task_id"] == TEST_TASK_ID
+    assert result_data["processed"] is True
+
+
+def test_async_activity_with_await():
+    """Validates async activity that performs async I/O"""
+
+    async def async_activity_with_io(ctx: task.ActivityContext, delay: float):
+        # Simulate async I/O operation
+        await asyncio.sleep(delay)
+        return f"completed after {delay}s"
+
+    activity_input = 0.01  # 10ms delay
+    executor, name = _get_activity_executor(async_activity_with_io)
+    result = asyncio.run(
+        executor.execute(TEST_INSTANCE_ID, name, TEST_TASK_ID, json.dumps(activity_input))
+    )
+    assert result is not None
+
+    result_output = json.loads(result)
+    assert result_output == "completed after 0.01s"
+
+
+def test_mixed_sync_async_activities():
+    """Validates that sync and async activities work together"""
+
+    def sync_activity(ctx: task.ActivityContext, test_input: str):
+        return f"sync:{test_input}"
+
+    async def async_activity(ctx: task.ActivityContext, test_input: str):
+        return f"async:{test_input}"
+
+    registry = worker._Registry()
+    sync_name = registry.add_activity(sync_activity)
+    async_name = registry.add_activity(async_activity)
+    executor = worker._ActivityExecutor(registry, TEST_LOGGER)
+
+    activity_input = "test"
+
+    # Execute sync activity
+    sync_result = asyncio.run(
+        executor.execute(TEST_INSTANCE_ID, sync_name, TEST_TASK_ID, json.dumps(activity_input))
+    )
+    assert json.loads(sync_result) == "sync:test"
+
+    # Execute async activity
+    async_result = asyncio.run(
+        executor.execute(TEST_INSTANCE_ID, async_name, TEST_TASK_ID + 1, json.dumps(activity_input))
+    )
+    assert json.loads(async_result) == "async:test"