agent-framework-durabletask

import os from agent_framework_durabletask.orchestration import ( DurableTaskOrchestrator, DurableTaskOrchestrationContext, ) from agent_framework_durabletask.client import DurableTaskClient from agent_framework_durabletask.azure import AzureDurableTaskClient # 1. Define a Durable Task Orchestrator class MySimpleOrchestrator(DurableTaskOrchestrator): """ A simple orchestrator that logs its input and returns a processed string. In a real scenario, this would coordinate calls to 'activities'. """ async def orchestrate(self, context: DurableTaskOrchestrationContext, input_data: str): print(f"Orchestrator '{context.instance_id}' received input: '{input_data}'") # Simulate some async work by returning immediately for this example return f"Processed '{input_data}' at {context.current_utc_datetime}" # 2. Instantiate a Durable Task Client (e.g., for Azure Durable Functions) # This client requires connection details to an Azure Storage Account # which is used by Azure Durable Functions to manage orchestration state. azure_storage_connection_string = os.environ.get("AZURE_STORAGE_CONNECTION_STRING", "") task_hub_name = os.environ.get("DURABLETASK_HUB_NAME", "DefaultTaskHub") if azure_storage_connection_string: print("Initializing AzureDurableTaskClient...") try: client = AzureDurableTaskClient( task_hub_name=task_hub_name, azure_storage_connection_string=azure_storage_connection_string ) print(f"AzureDurableTaskClient initialized for task hub '{task_hub_name}'.") # In a real application, you would then use `client` to start and manage orchestrations: # import asyncio # async def run_orchestration(): # instance_id = await client.start_orchestration(MySimpleOrchestrator, "Hello DurableTask!") # print(f"Started orchestration with ID: {instance_id}") # # You'd typically poll or wait for the orchestration to complete # # status = await client.get_orchestration_status(instance_id) # # while status.runtime_status not in [OrchestrationRuntimeStatus.Completed, OrchestrationRuntimeStatus.Failed, OrchestrationRuntimeStatus.Terminated]: # # await asyncio.sleep(5) # # status = await client.get_orchestration_status(instance_id) # # print(f"Orchestration {instance_id} finished with status: {status.runtime_status}") # asyncio.run(run_orchestration()) except Exception as e: print(f"Failed to initialize AzureDurableTaskClient: {e}") print("Please ensure AZURE_STORAGE_CONNECTION_STRING is valid and points to an Azure Storage Account.") else: print("AZURE_STORAGE_CONNECTION_STRING environment variable not set.") print("Cannot initialize AzureDurableTaskClient without storage connection details.") print("To run this, ensure you have an Azure Storage Account and set its connection string.") print("\n--- Example finished. This code defines an orchestrator and attempts to initialize a client. ---") print("To execute orchestrations, you need a Durable Task backend host (e.g., Azure Functions).")