docs(samples): add standalone durabletask workflow and HITL samples

Add two samples under samples/04-hosting/durabletask demonstrating MAF workflows on a standalone Durable Task worker (no Azure Functions):

- 08_workflow: conditional spam-detection workflow started via DurableWorkflowClient.start_workflow / await_workflow_output.

- 09_workflow_hitl: content-moderation workflow that pauses with ctx.request_info and is resumed via DurableWorkflowClient.get_pending_hitl_requests / send_hitl_response.

Also add the durabletask workflow integration test (test_08_dt_workflow).

Author: ahmedmuhsin

python/packages/durabletask/tests/integration_tests/test_08_dt_workflow.py

@@ -0,0 +1,78 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Integration tests for the standalone durabletask workflow sample (08_workflow).
+
+Exercises the standalone (non-Azure-Functions) workflow path:
+- ``DurableAIAgentWorker.configure_workflow`` auto-registers the agent entities,
+  non-agent executor activities, and the workflow orchestrator.
+- A client starts the workflow by scheduling ``WORKFLOW_ORCHESTRATOR_NAME``.
+- Conditional routing sends spam to a non-agent handler and legitimate email
+  through a second agent and a sender executor.
+"""
+
+import logging
+
+import pytest
+from durabletask.client import OrchestrationStatus
+
+from agent_framework_durabletask import WORKFLOW_ORCHESTRATOR_NAME
+
+logging.basicConfig(level=logging.WARNING)
+
+# Module-level markers
+pytestmark = [
+    pytest.mark.flaky,
+    pytest.mark.integration,
+    pytest.mark.sample("08_workflow"),
+    pytest.mark.integration_test,
+    pytest.mark.requires_dts,
+]
+
+
+class TestStandaloneWorkflow:
+    """Standalone (non-Azure-Functions) workflow execution on a durabletask worker."""
+
+    @pytest.fixture(autouse=True)
+    def setup(self, agent_client_factory: type, orchestration_helper) -> None:
+        """Provide a DTS client and orchestration helper for each test."""
+        self.dts_client, self.agent_client = agent_client_factory.create()
+        self.orch_helper = orchestration_helper
+
+    def test_legitimate_email_drafts_response(self) -> None:
+        """A legitimate email routes through the email agent and is 'sent'."""
+        instance_id = self.dts_client.schedule_new_orchestration(
+            orchestrator=WORKFLOW_ORCHESTRATOR_NAME,
+            input=(
+                "Hi team, just a reminder about our sprint planning meeting tomorrow at 10 AM. "
+                "Please review the agenda in Jira."
+            ),
+        )
+
+        metadata, output = self.orch_helper.wait_for_orchestration_with_output(
+            instance_id=instance_id,
+            timeout=180.0,
+        )
+
+        assert metadata.runtime_status == OrchestrationStatus.COMPLETED
+        assert output is not None
+        assert "Email sent" in str(output)
+
+    def test_spam_email_handled(self) -> None:
+        """A spam email routes to the non-agent spam handler."""
+        instance_id = self.dts_client.schedule_new_orchestration(
+            orchestrator=WORKFLOW_ORCHESTRATOR_NAME,
+            input="URGENT! You've won $1,000,000! Click here now to claim your prize! Limited time offer!",
+        )
+
+        metadata, output = self.orch_helper.wait_for_orchestration_with_output(
+            instance_id=instance_id,
+            timeout=180.0,
+        )
+
+        assert metadata.runtime_status == OrchestrationStatus.COMPLETED
+        assert output is not None
+        assert "spam" in str(output).lower()
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])

python/samples/04-hosting/durabletask/08_workflow/README.md

@@ -0,0 +1,57 @@
+# Workflow on a Standalone Durable Task Worker
+
+This sample demonstrates running an agent-framework `Workflow` as a durable
+orchestration on a **standalone Durable Task worker** — no Azure Functions
+required. It is the durabletask counterpart to the Azure Functions workflow
+samples (`samples/04-hosting/azure_functions/10_workflow_no_shared_state`).
+
+## Key Concepts Demonstrated
+
+- Hosting a MAF `Workflow` outside Azure Functions via
+  `DurableAIAgentWorker.configure_workflow(workflow)`, which auto-registers:
+  - a durable **entity** for each agent executor,
+  - a durable **activity** for each non-agent executor, and
+  - the **workflow orchestrator** (registered as `WORKFLOW_ORCHESTRATOR_NAME`).
+- Conditional routing with `add_switch_case_edge_group` (spam vs. legitimate email).
+- Mixing AI agents with non-agent executors in one workflow graph.
+- Starting the workflow from a client with
+  `DurableWorkflowClient.start_workflow(input=...)` and reading its result with
+  `await_workflow_output(instance_id)`.
+
+## Environment Setup
+
+See the [README.md](../README.md) in the parent directory for environment setup.
+
+This sample uses Azure AI Foundry credentials:
+
+- `FOUNDRY_PROJECT_ENDPOINT`
+- `FOUNDRY_MODEL`
+
+It also needs a Durable Task Scheduler. For local development, start the
+emulator (defaults to `http://localhost:8080`):
+
+```bash
+docker run -d -p 8080:8080 -p 8082:8082 mcr.microsoft.com/dts/dts-emulator:latest
+```
+
+## Running the Sample
+
+Start the worker in one terminal:
+
+```bash
+cd samples/04-hosting/durabletask/08_workflow
+python worker.py
+```
+
+In a second terminal, run the client:
+
+```bash
+python client.py
+```
+
+The client runs two cases:
+
+- **Legitimate email** → `SpamDetectionAgent` → `EmailAssistantAgent` →
+  `email_sender` → `"Email sent: ..."`.
+- **Spam email** → `SpamDetectionAgent` → `spam_handler` →
+  `"Email marked as spam: ..."`.

python/samples/04-hosting/durabletask/08_workflow/client.py

@@ -0,0 +1,73 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Client that starts the standalone workflow orchestration and prints the result.
+
+The worker (``worker.py``) must be running first. The workflow is started via
+``DurableAIAgentClient.start_workflow`` - which schedules the orchestrator that
+``DurableAIAgentWorker.configure_workflow`` auto-registers, so the caller never
+needs to know its internal name.
+
+Prerequisites:
+- ``worker.py`` running and connected to the same Durable Task Scheduler.
+- A Durable Task Scheduler reachable at ``ENDPOINT`` (default ``http://localhost:8080``).
+"""
+
+import asyncio
+import logging
+import os
+
+from agent_framework.azure import DurableWorkflowClient
+from azure.identity import AzureCliCredential
+from dotenv import load_dotenv
+from durabletask.azuremanaged.client import DurableTaskSchedulerClient
+
+load_dotenv()
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+def get_client(taskhub: str | None = None, endpoint: str | None = None) -> DurableTaskSchedulerClient:
+    """Create a configured DurableTaskSchedulerClient."""
+    taskhub_name = taskhub or os.getenv("TASKHUB", "default")
+    endpoint_url = endpoint or os.getenv("ENDPOINT", "http://localhost:8080")
+
+    credential = None if endpoint_url == "http://localhost:8080" else AzureCliCredential()
+
+    return DurableTaskSchedulerClient(
+        host_address=endpoint_url,
+        secure_channel=endpoint_url != "http://localhost:8080",
+        taskhub=taskhub_name,
+        token_credential=credential,
+    )
+
+
+def run_workflow(client: DurableWorkflowClient, email_content: str) -> None:
+    """Start the workflow with an email and wait for the result."""
+    instance_id = client.start_workflow(input=email_content)
+    logger.info("Started workflow instance: %s", instance_id)
+
+    output = client.await_workflow_output(instance_id)
+    logger.info("Workflow output: %s", output)
+
+
+async def main() -> None:
+    """Run the workflow against a legitimate email and a spam email."""
+    client = DurableWorkflowClient(get_client())
+
+    logger.info("TEST 1: Legitimate email")
+    run_workflow(
+        client,
+        "Hi team, just a reminder about our sprint planning meeting tomorrow at 10 AM. "
+        "Please review the agenda in Jira.",
+    )
+
+    logger.info("TEST 2: Spam email")
+    run_workflow(
+        client,
+        "URGENT! You've won $1,000,000! Click here now to claim your prize! Limited time offer!",
+    )
+
+
+if __name__ == "__main__":
+    asyncio.run(main())