Merge branch 'durable-openai-agent' into philliphoff-more-serialization

Author: philliphoff

azure/durable_functions/openai_agents/context.py

@@ -1,4 +1,5 @@
-from typing import Any, Callable, Optional
+import json
+from typing import Any, Callable, Optional, TYPE_CHECKING, Union
 
 from azure.durable_functions.models.DurableOrchestrationContext import (
     DurableOrchestrationContext,
@@ -8,11 +9,44 @@
 from agents import RunContextWrapper, Tool
 from agents.function_schema import function_schema
 from agents.tool import FunctionTool
+
+from azure.durable_functions.models.Task import TaskBase
 from .task_tracker import TaskTracker
 
 
-class DurableAIAgentContext:
-    """Context for AI agents running in Azure Durable Functions orchestration."""
+if TYPE_CHECKING:
+    # At type-check time we want all members / signatures for IDE & linters.
+    _BaseDurableContext = DurableOrchestrationContext
+else:
+    class _BaseDurableContext:  # lightweight runtime stub
+        """Runtime stub base class for delegation; real context is wrapped.
+
+        At runtime we avoid inheriting from DurableOrchestrationContext so that
+        attribute lookups for its members are delegated via __getattr__ to the
+        wrapped ``_context`` instance.
+        """
+
+        __slots__ = ()
+
+
+class DurableAIAgentContext(_BaseDurableContext):
+    """Context for AI agents running in Azure Durable Functions orchestration.
+
+    Design
+    ------
+    * Static analysis / IDEs: Appears to subclass ``DurableOrchestrationContext`` so
+      you get autocompletion and type hints (under TYPE_CHECKING branch).
+    * Runtime: Inherits from a trivial stub. All durable orchestration operations
+      are delegated to the real ``DurableOrchestrationContext`` instance provided
+      as ``context`` and stored in ``_context``.
+
+    Consequences
+    ------------
+    * ``isinstance(DurableAIAgentContext, DurableOrchestrationContext)`` is **False** at
+      runtime (expected).
+    * Delegation via ``__getattr__`` works for every member of the real context.
+    * No reliance on internal initialization side-effects of the durable SDK.
+    """
 
     def __init__(
         self,
@@ -24,28 +58,56 @@ def __init__(
         self._task_tracker = task_tracker
         self._model_retry_options = model_retry_options
 
-    def call_activity(self, activity_name, input: str):
-        """Call an activity function and record the activity call."""
-        task = self._context.call_activity(activity_name, input)
+    def call_activity(
+        self, name: Union[str, Callable], input_: Optional[Any] = None
+    ) -> TaskBase:
+        """Schedule an activity for execution.
+
+        Parameters
+        ----------
+        name: str | Callable
+            Either the name of the activity function to call, as a string or,
+            in the Python V2 programming model, the activity function itself.
+        input_: Optional[Any]
+            The JSON-serializable input to pass to the activity function.
+
+        Returns
+        -------
+        Task
+            A Durable Task that completes when the called activity function completes or fails.
+        """
+        task = self._context.call_activity(name, input_)
         self._task_tracker.record_activity_call()
         return task
 
     def call_activity_with_retry(
-        self, activity_name, retry_options: RetryOptions, input: str = None
-    ):
-        """Call an activity function with retry options and record the activity call."""
-        task = self._context.call_activity_with_retry(activity_name, retry_options, input)
+        self,
+        name: Union[str, Callable],
+        retry_options: RetryOptions,
+        input_: Optional[Any] = None,
+    ) -> TaskBase:
+        """Schedule an activity for execution with retry options.
+
+        Parameters
+        ----------
+        name: str | Callable
+            Either the name of the activity function to call, as a string or,
+            in the Python V2 programming model, the activity function itself.
+        retry_options: RetryOptions
+            The retry options for the activity function.
+        input_: Optional[Any]
+            The JSON-serializable input to pass to the activity function.
+
+        Returns
+        -------
+        Task
+            A Durable Task that completes when the called activity function completes or
+            fails completely.
+        """
+        task = self._context.call_activity_with_retry(name, retry_options, input_)
         self._task_tracker.record_activity_call()
         return task
 
-    def set_custom_status(self, status: str):
-        """Set custom status for the orchestration."""
-        self._context.set_custom_status(status)
-
-    def wait_for_external_event(self, event_name: str):
-        """Wait for an external event in the orchestration."""
-        return self._context.wait_for_external_event(event_name)
-
     def create_activity_tool(
         self,
         activity_func: Callable,
@@ -77,13 +139,29 @@ def create_activity_tool(
         else:
             activity_name = activity_func._function._name
 
+        input_name = None
+        if (activity_func._function._trigger is not None
+                and hasattr(activity_func._function._trigger, 'name')):
+            input_name = activity_func._function._trigger.name
+
         async def run_activity(ctx: RunContextWrapper[Any], input: str) -> Any:
+            # Parse JSON input and extract the named value if input_name is specified
+            activity_input = input
+            if input_name:
+                try:
+                    parsed_input = json.loads(input)
+                    if isinstance(parsed_input, dict) and input_name in parsed_input:
+                        activity_input = parsed_input[input_name]
+                    # If parsing fails or the named parameter is not found, pass the original input
+                except (json.JSONDecodeError, TypeError):
+                    pass
+
             if retry_options:
                 result = self._task_tracker.get_activity_call_result_with_retry(
-                    activity_name, retry_options, input
+                    activity_name, retry_options, activity_input
                 )
             else:
-                result = self._task_tracker.get_activity_call_result(activity_name, input)
+                result = self._task_tracker.get_activity_call_result(activity_name, activity_input)
             return result
 
         schema = function_schema(
@@ -101,3 +179,14 @@ async def run_activity(ctx: RunContextWrapper[Any], input: str) -> Any:
             on_invoke_tool=run_activity,
             strict_json_schema=True,
         )
+
+    def __getattr__(self, name):
+        """Delegate missing attributes to the underlying DurableOrchestrationContext."""
+        try:
+            return getattr(self._context, name)
+        except AttributeError:
+            raise AttributeError(f"'{type(self).__name__}' object has no attribute '{name}'")
+
+    def __dir__(self):
+        """Improve introspection and tab-completion by including delegated attributes."""
+        return sorted(set(dir(type(self)) + list(self.__dict__) + dir(self._context)))

azure/durable_functions/openai_agents/model_invocation_activity.py

@@ -163,7 +163,16 @@ class ActivityModelInput(BaseModel):
 
     def to_json(self) -> str:
         """Convert the ActivityModelInput to a JSON string."""
-        return self.model_dump_json()
+        try:
+            return self.model_dump_json(warnings=False)
+        except Exception:
+            # Fallback to basic JSON serialization
+            try:
+                return json.dumps(self.model_dump(warnings=False), default=str)
+            except Exception as fallback_error:
+                raise ValueError(
+                    f"Unable to serialize ActivityModelInput: {fallback_error}"
+                ) from fallback_error
 
     @classmethod
     def from_json(cls, json_str: str) -> 'ActivityModelInput':
@@ -286,6 +295,7 @@ async def get_response(
         *,
         previous_response_id: Optional[str],
         prompt: Optional[ResponsePromptParam],
+        conversation_id: Optional[str] = None,
     ) -> ModelResponse:
         """Get a response from the model."""
         def make_tool_info(tool: Tool) -> ToolInput:

azure/durable_functions/openai_agents/orchestrator_generator.py

@@ -9,6 +9,7 @@
 from .runner import DurableOpenAIRunner
 from .context import DurableAIAgentContext
 from .event_loop import ensure_event_loop
+from .usage_telemetry import UsageTelemetry
 
 
 async def durable_openai_agent_activity(input: str, model_provider: ModelProvider) -> str:
@@ -18,9 +19,27 @@ async def durable_openai_agent_activity(input: str, model_provider: ModelProvide
     model_invoker = ModelInvoker(model_provider=model_provider)
     response = await model_invoker.invoke_model_activity(activity_input)
 
-    activity_output = ActivityModelOutput(response=response)
+    # Use safe/public Pydantic API when possible. Prefer model_dump_json if result is a BaseModel
+    # Otherwise handle common types (str/bytes/dict/list) and fall back to json.dumps.
+    import json as _json
 
-    return activity_output.to_json()
+    if hasattr(result, "model_dump_json"):
+        # Pydantic v2 BaseModel
+        json_str = result.model_dump_json()
+    else:
+        if isinstance(result, bytes):
+            json_str = result.decode()
+        elif isinstance(result, str):
+            json_str = result
+        else:
+            # Try the internal serializer as a last resort, but fall back to json.dumps
+            try:
+                json_bytes = ModelResponse.__pydantic_serializer__.to_json(result)
+                json_str = json_bytes.decode()
+            except Exception:
+                json_str = _json.dumps(result)
+
+    return json_str
 
 
 def durable_openai_agent_orchestrator_generator(
@@ -30,6 +49,9 @@ def durable_openai_agent_orchestrator_generator(
         activity_name: str,
 ):
     """Adapts the synchronous OpenAI Agents function to an Durable orchestrator generator."""
+    # Log versions the first time this generator is invoked
+    UsageTelemetry.log_usage_once()
+
     ensure_event_loop()
     task_tracker = TaskTracker(durable_orchestration_context)
     durable_ai_agent_context = DurableAIAgentContext(

azure/durable_functions/openai_agents/task_tracker.py

@@ -52,13 +52,13 @@ def _get_activity_result_or_raise(self, task):
         result = json.loads(result_json)
         return result
 
-    def get_activity_call_result(self, activity_name, input: str):
+    def get_activity_call_result(self, activity_name, input: Any):
         """Call an activity and return its result or raise ``YieldException`` if pending."""
         task = self._context.call_activity(activity_name, input)
         return self._get_activity_result_or_raise(task)
 
     def get_activity_call_result_with_retry(
-        self, activity_name, retry_options: RetryOptions, input: str
+        self, activity_name, retry_options: RetryOptions, input: Any
     ):
         """Call an activity with retry and return its result or raise YieldException if pending."""
         task = self._context.call_activity_with_retry(activity_name, retry_options, input)

azure/durable_functions/openai_agents/usage_telemetry.py

@@ -0,0 +1,69 @@
+#  Copyright (c) Microsoft Corporation. All rights reserved.
+#  Licensed under the MIT License.
+
+
+class UsageTelemetry:
+    """Handles telemetry logging for OpenAI Agents SDK integration usage."""
+
+    # Class-level flag to ensure logging happens only once across all instances
+    _usage_logged = False
+
+    @classmethod
+    def log_usage_once(cls):
+        """Log OpenAI Agents SDK integration usage exactly once.
+
+        Fails gracefully if metadata cannot be retrieved.
+        """
+        if cls._usage_logged:
+            return
+
+        # NOTE: Any log line beginning with the special prefix defined below will be
+        # captured by the Azure Functions host as a Language Worker console log and
+        # forwarded to internal telemetry pipelines.
+        # Do not change this constant value without coordinating with the Functions
+        # host team.
+        LANGUAGE_WORKER_CONSOLE_LOG_PREFIX = "LanguageWorkerConsoleLog"
+
+        package_versions = cls._collect_openai_agent_package_versions()
+        msg = (
+            f"{LANGUAGE_WORKER_CONSOLE_LOG_PREFIX}"  # Prefix captured by Azure Functions host
+            "Detected OpenAI Agents SDK integration with Durable Functions. "
+            f"Package versions: {package_versions}"
+        )
+        print(msg)
+
+        cls._usage_logged = True
+
+    @classmethod
+    def _collect_openai_agent_package_versions(cls) -> str:
+        """Collect versions of relevant packages for telemetry logging.
+
+        Returns
+        -------
+        str
+            Comma-separated list of name=version entries or "(unavailable)" if
+            versions could not be determined.
+        """
+        try:
+            try:
+                from importlib import metadata  # Python 3.8+
+            except ImportError:  # pragma: no cover - legacy fallback
+                import importlib_metadata as metadata  # type: ignore
+
+            package_names = [
+                "azure-functions-durable",
+                "openai",
+                "openai-agents",
+            ]
+
+            versions = []
+            for package_name in package_names:
+                try:
+                    ver = metadata.version(package_name)
+                    versions.append(f"{package_name}={ver}")
+                except Exception:  # noqa: BLE001 - swallow and continue
+                    versions.append(f"{package_name}=(not installed)")
+
+            return ", ".join(versions) if versions else "(unavailable)"
+        except Exception:  # noqa: BLE001 - never let version gathering break user code
+            return "(unavailable)"

samples-v2/openai_agents/function_app.py

@@ -1,4 +1,5 @@
 import os
+import random
 
 import azure.functions as func
 import azure.durable_functions as df
@@ -102,4 +103,13 @@ def tools(context):
     import basic.tools
     return basic.tools.main()
 
+@app.activity_trigger(input_name="max")
+async def random_number_tool(max: int) -> int:
+    """Return a random integer between 0 and the given maximum."""
+    return random.randint(0, max)
 
+@app.orchestration_trigger(context_name="context")
+@app.durable_openai_agent_orchestrator
+def message_filter(context):
+    import handoffs.message_filter
+    return handoffs.message_filter.main(context.create_activity_tool(random_number_tool))