Move implementation to azure/durable_functions/openai_agents

Author: AnatoliB

azure/durable_functions/__init__.py

@@ -79,3 +79,11 @@ def validate_extension_bundles():
     __all__.append('Blueprint')
 except ModuleNotFoundError:
     pass
+
+# Import OpenAI Agents integration (optional dependency)
+try:
+    from . import openai_agents # noqa
+    __all__.append('openai_agents')
+except ImportError:
+    # OpenAI agents integration requires additional dependencies
+    pass

azure/durable_functions/openai_agents/__init__.py

@@ -0,0 +1,13 @@
+"""OpenAI Agents integration for Azure Durable Functions.
+
+This module provides decorators and utilities to integrate OpenAI Agents
+with Azure Durable Functions orchestration patterns.
+"""
+
+from .decorators import durable_openai_agent_orchestrator
+from .context import DurableAIAgentContext
+
+__all__ = [
+    'durable_openai_agent_orchestrator',
+    'DurableAIAgentContext', 
+]

samples-v2/openai_agents/durable_ai_agent_context.py renamed to azure/durable_functions/openai_agents/context.py

@@ -8,7 +8,7 @@
 from agents import RunContextWrapper, Tool
 from agents.function_schema import function_schema
 from agents.tool import FunctionTool
-from yield_exception import YieldException
+from .exceptions import YieldException
 
 
 class DurableAIAgentContext:
@@ -63,6 +63,7 @@ def activity_as_tool(
 
         Args:
             activity_func: The Azure Functions activity function to convert
+            description: Optional description override for the tool
 
         Returns:
             Tool: An OpenAI Agents SDK Tool object
@@ -89,4 +90,4 @@ async def run_activity(ctx: RunContextWrapper[Any], input: str) -> Any:
             params_json_schema=schema.params_json_schema,
             on_invoke_tool=run_activity,
             strict_json_schema=True,
-        )
+        )

samples-v2/openai_agents/durable_decorators.py renamed to azure/durable_functions/openai_agents/decorators.py

@@ -4,11 +4,11 @@
 import azure.functions as func
 from agents.run import set_default_agent_runner
 from azure.durable_functions.models.DurableOrchestrationContext import DurableOrchestrationContext
-from durable_openai_runner import DurableOpenAIRunner
-from yield_exception import YieldException
-from durable_ai_agent_context import DurableAIAgentContext
-import event_loop
-from model_activity import create_invoke_model_activity
+from .runner import DurableOpenAIRunner
+from .exceptions import YieldException
+from .context import DurableAIAgentContext
+from .event_loop import ensure_event_loop
+from .model_invocation_activity import create_invoke_model_activity
 
 
 # Global registry to track which apps have been set up
@@ -67,7 +67,7 @@ def durable_openai_agent_orchestrator(func):
 
     @wraps(func)
     def wrapper(durable_orchestration_context: DurableOrchestrationContext):
-        event_loop.ensure_event_loop()
+        ensure_event_loop()
         durable_ai_agent_context = DurableAIAgentContext(durable_orchestration_context)
         durable_openai_runner = DurableOpenAIRunner(context=durable_ai_agent_context)
         set_default_agent_runner(durable_openai_runner)
@@ -116,4 +116,4 @@ def wrapper(durable_orchestration_context: DurableOrchestrationContext):
             finally:
                 yield from durable_ai_agent_context._yield_and_clear_tasks()
 
-    return wrapper
+    return wrapper

samples-v2/openai_agents/event_loop.py renamed to azure/durable_functions/openai_agents/event_loop.py

@@ -13,4 +13,4 @@ def ensure_event_loop():
         asyncio.get_running_loop()
     except RuntimeError:
         loop = asyncio.new_event_loop()
-        asyncio.set_event_loop(loop)
+        asyncio.set_event_loop(loop)

samples-v2/openai_agents/yield_exception.py renamed to azure/durable_functions/openai_agents/exceptions.py

@@ -3,4 +3,4 @@
 class YieldException(BaseException):
     def __init__(self, task: TaskBase):
         super().__init__("Orchestrator should yield.")
-        self.task = task
+        self.task = task

samples-v2/openai_agents/model_invoker.py renamed to azure/durable_functions/openai_agents/model_invocation_activity.py

@@ -1,18 +1,23 @@
+from __future__ import annotations
 import enum
 import json
+import logging
 from dataclasses import dataclass
 from datetime import timedelta
-from typing import Any, Optional, Union
+from typing import Any, AsyncIterator, Optional, Union, cast
 
+import azure.functions as func
 from pydantic import BaseModel, Field
 from agents import (
+    AgentOutputSchema,
     AgentOutputSchemaBase,
     CodeInterpreterTool,
     FileSearchTool,
     FunctionTool,
     Handoff,
     HostedMCPTool,
     ImageGenerationTool,
+    Model,
     ModelProvider,
     ModelResponse,
     ModelSettings,
@@ -24,11 +29,16 @@
     UserError,
     WebSearchTool,
 )
+from agents.items import TResponseStreamEvent
 from openai import (
     APIStatusError,
     AsyncOpenAI,
 )
 from openai.types.responses.tool_param import Mcp
+from openai.types.responses.response_prompt_param import ResponsePromptParam
+
+from .context import DurableAIAgentContext
+
 try:
     from azure.durable_functions import ApplicationError
 except ImportError:
@@ -39,6 +49,8 @@ def __init__(self, message: str, non_retryable: bool = False, next_retry_delay =
             self.non_retryable = non_retryable
             self.next_retry_delay = next_retry_delay
 
+logger = logging.getLogger(__name__)
+
 
 class HandoffInput(BaseModel):
     """Data conversion friendly representation of a Handoff. Contains only the fields which are needed by the model
@@ -259,3 +271,132 @@ def make_tool(tool: ToolInput) -> Tool:
                 non_retryable=True,
                 next_retry_delay=retry_after,
             ) from e
+
+
+class _DurableModelStub(Model):
+    def __init__(
+        self,
+        model_name: Optional[str],
+        context: DurableAIAgentContext,
+    ) -> None:
+        self.model_name = model_name
+        self.context = context
+
+    async def get_response(
+        self,
+        system_instructions: Optional[str],
+        input: Union[str, list[TResponseInputItem]],
+        model_settings: ModelSettings,
+        tools: list[Tool],
+        output_schema: Optional[AgentOutputSchemaBase],
+        handoffs: list[Handoff],
+        tracing: ModelTracing,
+        *,
+        previous_response_id: Optional[str],
+        prompt: Optional[ResponsePromptParam],
+    ) -> ModelResponse:
+        def make_tool_info(tool: Tool) -> ToolInput:
+            if isinstance(
+                tool,
+                (
+                    FileSearchTool,
+                    WebSearchTool,
+                    ImageGenerationTool,
+                    CodeInterpreterTool,
+                ),
+            ):
+                return tool
+            elif isinstance(tool, HostedMCPTool):
+                return HostedMCPToolInput(tool_config=tool.tool_config)
+            elif isinstance(tool, FunctionTool):
+                return FunctionToolInput(
+                    name=tool.name,
+                    description=tool.description,
+                    params_json_schema=tool.params_json_schema,
+                    strict_json_schema=tool.strict_json_schema,
+                )
+            else:
+                raise ValueError(f"Unsupported tool type: {tool.name}")
+
+        tool_infos = [make_tool_info(x) for x in tools]
+        handoff_infos = [
+            HandoffInput(
+                tool_name=x.tool_name,
+                tool_description=x.tool_description,
+                input_json_schema=x.input_json_schema,
+                agent_name=x.agent_name,
+                strict_json_schema=x.strict_json_schema,
+            )
+            for x in handoffs
+        ]
+        if output_schema is not None and not isinstance(
+            output_schema, AgentOutputSchema
+        ):
+            raise TypeError(
+                f"Only AgentOutputSchema is supported by Durable Model, got {type(output_schema).__name__}"
+            )
+        agent_output_schema = output_schema
+        output_schema_input = (
+            None
+            if agent_output_schema is None
+            else AgentOutputSchemaInput(
+                output_type_name=agent_output_schema.name(),
+                is_wrapped=agent_output_schema._is_wrapped,
+                output_schema=agent_output_schema.json_schema()
+                if not agent_output_schema.is_plain_text()
+                else None,
+                strict_json_schema=agent_output_schema.is_strict_json_schema(),
+            )
+        )
+
+        activity_input = ActivityModelInput(
+            model_name=self.model_name,
+            system_instructions=system_instructions,
+            input=cast(Union[str, list[TResponseInputItem]], input),
+            model_settings=model_settings,
+            tools=tool_infos,
+            output_schema=output_schema_input,
+            handoffs=handoff_infos,
+            tracing=ModelTracingInput.DISABLED,  # ModelTracingInput(tracing.value),
+            previous_response_id=previous_response_id,
+            prompt=prompt,
+        )
+
+        activity_input_json = activity_input.to_json()
+
+        response = self.context._get_activity_call_result("invoke_model_activity", activity_input_json)
+        json_response = json.loads(response)
+        model_response = ModelResponse(**json_response)
+        return model_response
+
+    def stream_response(
+        self,
+        system_instructions: Optional[str],
+        input: Union[str, list[TResponseInputItem]],
+        model_settings: ModelSettings,
+        tools: list[Tool],
+        output_schema: Optional[AgentOutputSchemaBase],
+        handoffs: list[Handoff],
+        tracing: ModelTracing,
+        *,
+        previous_response_id: Optional[str],
+        prompt: ResponsePromptParam | None,
+    ) -> AsyncIterator[TResponseStreamEvent]:
+        raise NotImplementedError("Durable model doesn't support streams yet")
+
+
+def create_invoke_model_activity(app: func.FunctionApp):
+    """Create and register the invoke_model_activity function with the provided FunctionApp."""
+    
+    @app.activity_trigger(input_name="input")
+    async def invoke_model_activity(input: str):
+        """Activity that handles OpenAI model invocations."""
+        activity_input = ActivityModelInput.from_json(input)
+        
+        model_invoker = ModelInvoker()
+        result = await model_invoker.invoke_model_activity(activity_input)
+
+        json_obj = ModelResponse.__pydantic_serializer__.to_json(result)
+        return json_obj.decode()
+    
+    return invoke_model_activity

samples-v2/openai_agents/durable_openai_runner.py renamed to azure/durable_functions/openai_agents/runner.py

@@ -14,11 +14,12 @@
 from agents.run import DEFAULT_AGENT_RUNNER, DEFAULT_MAX_TURNS, AgentRunner
 from pydantic_core import to_json
 
-from durable_ai_agent_context import DurableAIAgentContext
-from durable_model_stub import _DurableModelStub
+from .context import DurableAIAgentContext
+from .model_invocation_activity import _DurableModelStub
 
 logger = logging.getLogger(__name__)
 
+
 class DurableOpenAIRunner:
     def __init__(self, context: DurableAIAgentContext) -> None:
         self._runner = DEFAULT_AGENT_RUNNER or AgentRunner()
@@ -84,4 +85,4 @@ def run_streamed(
         input: Union[str, list[TResponseInputItem]],
         **kwargs: Any,
     ) -> RunResultStreaming:
-        raise RuntimeError("Durable Functions do not support streaming.")
+        raise RuntimeError("Durable Functions do not support streaming.")