Python: Add Durabletask samples and minor fixes (#3157)

* Add samples and minor fixes

* Add redis sample and wait-for-completion

* Add wait-for-completion support

* ADd missing docs

python/packages/durabletask/agent_framework_durabletask/_entities.py

@@ -142,7 +142,7 @@ class AgentEntity:
         response_format = run_request.response_format
         enable_tool_calls = run_request.enable_tool_calls
 
-        logger.debug("[AgentEntity.run] Received Message: %s", run_request)
+        logger.debug("[AgentEntity.run] Received ThreadId %s Message: %s", thread_id, run_request)
 
         state_request = DurableAgentStateRequest.from_run_request(run_request)
         self.state.data.conversation_history.append(state_request)

python/packages/durabletask/agent_framework_durabletask/_executors.py

@@ -16,10 +16,10 @@ from abc import ABC, abstractmethod
 from datetime import datetime, timezone
 from typing import Any, Generic, TypeVar
 
-from agent_framework import AgentRunResponse, AgentThread, ChatMessage, ErrorContent, Role, get_logger
+from agent_framework import AgentRunResponse, AgentThread, ChatMessage, ErrorContent, Role, TextContent, get_logger
 from durabletask.client import TaskHubGrpcClient
 from durabletask.entities import EntityInstanceId
-from durabletask.task import CompositeTask, OrchestrationContext, Task
+from durabletask.task import CompletableTask, CompositeTask, OrchestrationContext, Task
 from pydantic import BaseModel
 
 from ._constants import DEFAULT_MAX_POLL_RETRIES, DEFAULT_POLL_INTERVAL_SECONDS
@@ -33,16 +33,19 @@ logger = get_logger("agent_framework.durabletask.executors")
 TaskT = TypeVar("TaskT")
 
 
-class DurableAgentTask(CompositeTask[AgentRunResponse]):
+class DurableAgentTask(CompositeTask[AgentRunResponse], CompletableTask[AgentRunResponse]):
     """A custom Task that wraps entity calls and provides typed AgentRunResponse results.
 
     This task wraps the underlying entity call task and intercepts its completion
     to convert the raw result into a typed AgentRunResponse object.
+
+    When yielded in an orchestration, this task returns an AgentRunResponse:
+        response: AgentRunResponse = yield durable_agent_task
     """
 
     def __init__(
         self,
-        entity_task: Task[Any],
+        entity_task: CompletableTask[Any],
         response_format: type[BaseModel] | None,
         correlation_id: str,
     ):
@@ -55,7 +58,7 @@ class DurableAgentTask(CompositeTask[AgentRunResponse]):
         """
         self._response_format = response_format
         self._correlation_id = correlation_id
-        super().__init__([entity_task])  # type: ignore[misc]
+        super().__init__([entity_task])  # type: ignore
 
     def on_child_completed(self, task: Task[Any]) -> None:
         """Handle completion of the underlying entity task.
@@ -69,11 +72,8 @@ class DurableAgentTask(CompositeTask[AgentRunResponse]):
             return
 
         if task.is_failed:
-            # Propagate the failure
-            self._exception = task.get_exception()
-            self._is_complete = True
-            if self._parent is not None:
-                self._parent.on_child_completed(self)
+            # Propagate the failure - pass the original exception directly
+            self.fail("call_entity Task failed", task.get_exception())
             return
 
         # Task succeeded - transform the raw result
@@ -94,18 +94,12 @@ class DurableAgentTask(CompositeTask[AgentRunResponse]):
                 )
 
             # Set the typed AgentRunResponse as this task's result
-            self._result = response
-            self._is_complete = True
+            self.complete(response)
 
-            if self._parent is not None:
-                self._parent.on_child_completed(self)
-
-        except Exception:
-            logger.exception(
-                "[DurableAgentTask] Failed to convert result for correlation_id: %s",
-                self._correlation_id,
-            )
-            raise
+        except Exception as ex:
+            err_msg = "[DurableAgentTask] Failed to convert result for correlation_id: " + self._correlation_id
+            logger.exception(err_msg)
+            self.fail(err_msg, ex)
 
 
 class DurableAgentExecutor(ABC, Generic[TaskT]):
@@ -155,6 +149,7 @@ class DurableAgentExecutor(ABC, Generic[TaskT]):
         message: str,
         response_format: type[BaseModel] | None,
         enable_tool_calls: bool,
+        wait_for_response: bool = True,
     ) -> RunRequest:
         """Create a RunRequest for the given parameters."""
         correlation_id = self.generate_unique_id()
@@ -162,9 +157,34 @@ class DurableAgentExecutor(ABC, Generic[TaskT]):
             message=message,
             response_format=response_format,
             enable_tool_calls=enable_tool_calls,
+            wait_for_response=wait_for_response,
             correlation_id=correlation_id,
         )
 
+    def _create_acceptance_response(self, correlation_id: str) -> AgentRunResponse:
+        """Create an acceptance response for fire-and-forget mode.
+
+        Args:
+            correlation_id: Correlation ID for tracking the request
+
+        Returns:
+            AgentRunResponse: Acceptance response with correlation ID
+        """
+        acceptance_message = ChatMessage(
+            role=Role.SYSTEM,
+            contents=[
+                TextContent(
+                    f"Request accepted for processing (correlation_id: {correlation_id}). "
+                    f"Agent is executing in the background. "
+                    f"Retrieve response via your configured streaming or callback mechanism."
+                )
+            ],
+        )
+        return AgentRunResponse(
+            messages=[acceptance_message],
+            created_at=datetime.now(timezone.utc).isoformat(),
+        )
+
 
 class ClientAgentExecutor(DurableAgentExecutor[AgentRunResponse]):
     """Execution strategy for external clients.
@@ -205,11 +225,20 @@ class ClientAgentExecutor(DurableAgentExecutor[AgentRunResponse]):
             thread: Optional conversation thread (creates new if not provided)
 
         Returns:
-            AgentRunResponse: The agent's response after execution completes
+            AgentRunResponse: The agent's response after execution completes, or an immediate
+                            acknowledgement if wait_for_response is False
         """
         # Signal the entity with the request
         entity_id = self._signal_agent_entity(agent_name, run_request, thread)
 
+        # If fire-and-forget mode, return immediately without polling
+        if not run_request.wait_for_response:
+            logger.info(
+                "[ClientAgentExecutor] Fire-and-forget mode: request signaled (correlation: %s)",
+                run_request.correlation_id,
+            )
+            return self._create_acceptance_response(run_request.correlation_id)
+
         # Poll for the response
         agent_response = self._poll_for_agent_response(entity_id, run_request.correlation_id)
 
@@ -395,11 +424,16 @@ class OrchestrationAgentExecutor(DurableAgentExecutor[DurableAgentTask]):
         self._context = context
         logger.debug("[OrchestrationAgentExecutor] Initialized")
 
+    def generate_unique_id(self) -> str:
+        """Create a new UUID that is safe for replay within an orchestration or operation."""
+        return self._context.new_uuid()
+
     def get_run_request(
         self,
         message: str,
         response_format: type[BaseModel] | None,
         enable_tool_calls: bool,
+        wait_for_response: bool = True,
     ) -> RunRequest:
         """Get the current run request from the orchestration context.
 
@@ -410,6 +444,7 @@ class OrchestrationAgentExecutor(DurableAgentExecutor[DurableAgentTask]):
             message,
             response_format,
             enable_tool_calls,
+            wait_for_response,
         )
         request.orchestration_id = self._context.instance_id
         return request
@@ -449,8 +484,22 @@ class OrchestrationAgentExecutor(DurableAgentExecutor[DurableAgentTask]):
             session_id,
         )
 
-        # Call the entity and get the underlying task
-        entity_task: Task[Any] = self._context.call_entity(entity_id, "run", run_request.to_dict())  # type: ignore
+        # Branch based on wait_for_response
+        if not run_request.wait_for_response:
+            # Fire-and-forget mode: signal entity and return pre-completed task
+            logger.info(
+                "[OrchestrationAgentExecutor] Fire-and-forget mode: signaling entity (correlation: %s)",
+                run_request.correlation_id,
+            )
+            self._context.signal_entity(entity_id, "run", run_request.to_dict())
+
+            # Create a pre-completed task with acceptance response
+            acceptance_response = self._create_acceptance_response(run_request.correlation_id)
+            entity_task: CompletableTask[AgentRunResponse] = CompletableTask()
+            entity_task.complete(acceptance_response)
+        else:
+            # Blocking mode: call entity and wait for response
+            entity_task = self._context.call_entity(entity_id, "run", run_request.to_dict())  # type: ignore
 
         # Wrap in DurableAgentTask for response transformation
         return DurableAgentTask(

python/packages/durabletask/agent_framework_durabletask/_models.py

@@ -104,6 +104,8 @@ class RunRequest:
         role: The role of the message sender (user, system, or assistant)
         response_format: Optional Pydantic BaseModel type describing the structured response format
         enable_tool_calls: Whether to enable tool calls for this request
+        wait_for_response: If True (default), caller will wait for agent response. If False,
+                          returns immediately after signaling (fire-and-forget mode)
         correlation_id: Correlation ID for tracking the response to this specific request
         created_at: Optional timestamp when the request was created
         orchestration_id: Optional ID of the orchestration that initiated this request
@@ -115,6 +117,7 @@ class RunRequest:
     role: Role = Role.USER
     response_format: type[BaseModel] | None = None
     enable_tool_calls: bool = True
+    wait_for_response: bool = True
     created_at: datetime | None = None
     orchestration_id: str | None = None
 
@@ -126,6 +129,7 @@ class RunRequest:
         role: Role | str | None = Role.USER,
         response_format: type[BaseModel] | None = None,
         enable_tool_calls: bool = True,
+        wait_for_response: bool = True,
         created_at: datetime | None = None,
         orchestration_id: str | None = None,
     ) -> None:
@@ -135,6 +139,7 @@ class RunRequest:
         self.response_format = response_format
         self.request_response_format = request_response_format
         self.enable_tool_calls = enable_tool_calls
+        self.wait_for_response = wait_for_response
         self.created_at = created_at if created_at is not None else datetime.now(tz=timezone.utc)
         self.orchestration_id = orchestration_id
 
@@ -155,6 +160,7 @@ class RunRequest:
         result = {
             "message": self.message,
             "enable_tool_calls": self.enable_tool_calls,
+            "wait_for_response": self.wait_for_response,
             "role": self.role.value,
             "request_response_format": self.request_response_format,
             "correlationId": self.correlation_id,
@@ -198,6 +204,7 @@ class RunRequest:
             request_response_format=data.get("request_response_format", REQUEST_RESPONSE_FORMAT_TEXT),
             role=cls.coerce_role(data.get("role")),
             response_format=_deserialize_response_format(data.get("response_format")),
+            wait_for_response=data.get("wait_for_response", True),
             enable_tool_calls=data.get("enable_tool_calls", True),
             created_at=created_at,
             orchestration_id=data.get("orchestrationId"),

python/packages/durabletask/agent_framework_durabletask/_response_utils.py

@@ -51,10 +51,20 @@ def ensure_response_format(
         response_format: Optional Pydantic model class to parse the response value into
         correlation_id: Correlation ID for logging purposes
         response: The AgentRunResponse object to validate and parse
+
+    Raises:
+        ValueError: If response_format is specified but response.value cannot be parsed
     """
     if response_format is not None and not isinstance(response.value, response_format):
         response.try_parse_value(response_format)
 
+        # Validate that parsing succeeded
+        if not isinstance(response.value, response_format):
+            raise ValueError(
+                f"Response value could not be parsed into required format {response_format.__name__} "
+                f"for correlation_id {correlation_id}"
+            )
+
         logger.debug(
             "[ensure_response_format] Loaded AgentRunResponse.value for correlation_id %s with type: %s",
             correlation_id,

python/packages/durabletask/agent_framework_durabletask/_shim.py

@@ -108,9 +108,20 @@ class DurableAIAgent(AgentProtocol, Generic[TaskT]):
         thread: AgentThread | None = None,
         response_format: type[BaseModel] | None = None,
         enable_tool_calls: bool = True,
+        wait_for_response: bool = True,
     ) -> TaskT:
         """Execute the agent via the injected provider.
 
+        Args:
+            messages: The message(s) to send to the agent
+            thread: Optional agent thread for conversation context
+            response_format: Optional Pydantic model for structured response
+            enable_tool_calls: Whether to enable tool calls for this request
+            wait_for_response: If True (default), waits for agent response.
+                             If False, returns immediately (fire-and-forget mode).
+
+                             **Only supported for DurableAIAgentClient contexts.**
+
         Note:
             This method overrides AgentProtocol.run() with a different return type:
             - AgentProtocol.run() returns Coroutine[Any, Any, AgentRunResponse] (async)
@@ -121,6 +132,9 @@ class DurableAIAgent(AgentProtocol, Generic[TaskT]):
 
         Returns:
             TaskT: The task type specific to the executor
+
+        Raises:
+            ValueError: If wait_for_response=False is used in an unsupported context
         """
         message_str = self._normalize_messages(messages)
 
@@ -128,6 +142,7 @@ class DurableAIAgent(AgentProtocol, Generic[TaskT]):
             message=message_str,
             response_format=response_format,
             enable_tool_calls=enable_tool_calls,
+            wait_for_response=wait_for_response,
         )
 
         return self._executor.run_durable_agent(