Convert class to module (#958)

Author: dandavison

temporalio/contrib/openai_agents/__init__.py

@@ -17,9 +17,10 @@
     TestModel,
     TestModelProvider,
     set_open_ai_agent_temporal_overrides,
-    workflow,
 )
 
+from . import workflow
+
 __all__ = [
     "ModelActivity",
     "ModelActivityParameters",

temporalio/contrib/openai_agents/temporal_openai_agents.py

@@ -1,11 +1,8 @@
 """Initialize Temporal OpenAI Agents overrides."""
 
-import json
 from contextlib import contextmanager
-from datetime import timedelta
-from typing import Any, AsyncIterator, Callable, Optional, Type, Union
+from typing import AsyncIterator, Callable, Optional, Union
 
-import nexusrpc
 from agents import (
     AgentOutputSchemaBase,
     Handoff,
@@ -14,31 +11,21 @@
     ModelResponse,
     ModelSettings,
     ModelTracing,
-    RunContextWrapper,
     Tool,
     TResponseInputItem,
     set_trace_provider,
 )
-from agents.function_schema import function_schema
 from agents.items import TResponseStreamEvent
 from agents.run import get_default_agent_runner, set_default_agent_runner
-from agents.tool import (
-    FunctionTool,
-)
 from agents.tracing import get_trace_provider
 from agents.tracing.provider import DefaultTraceProvider
 from openai.types.responses import ResponsePromptParam
 
-from temporalio import activity
-from temporalio import workflow as temporal_workflow
-from temporalio.common import Priority, RetryPolicy
 from temporalio.contrib.openai_agents._model_parameters import ModelActivityParameters
 from temporalio.contrib.openai_agents._openai_runner import TemporalOpenAIRunner
 from temporalio.contrib.openai_agents._temporal_trace_provider import (
     TemporalTraceProvider,
 )
-from temporalio.exceptions import ApplicationError, TemporalError
-from temporalio.workflow import ActivityCancellationType, VersioningIntent
 
 
 @contextmanager
@@ -146,204 +133,3 @@ def stream_response(
     ) -> AsyncIterator[TResponseStreamEvent]:
         """Get a streamed response from the model. Unimplemented."""
         raise NotImplementedError()
-
-
-class ToolSerializationError(TemporalError):
-    """Error that occurs when a tool output could not be serialized."""
-
-
-class workflow:
-    """Encapsulates workflow specific primitives for working with the OpenAI Agents SDK in a workflow context"""
-
-    @classmethod
-    def activity_as_tool(
-        cls,
-        fn: Callable,
-        *,
-        task_queue: Optional[str] = None,
-        schedule_to_close_timeout: Optional[timedelta] = None,
-        schedule_to_start_timeout: Optional[timedelta] = None,
-        start_to_close_timeout: Optional[timedelta] = None,
-        heartbeat_timeout: Optional[timedelta] = None,
-        retry_policy: Optional[RetryPolicy] = None,
-        cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
-        activity_id: Optional[str] = None,
-        versioning_intent: Optional[VersioningIntent] = None,
-        summary: Optional[str] = None,
-        priority: Priority = Priority.default,
-    ) -> Tool:
-        """Convert a single Temporal activity function to an OpenAI agent tool.
-
-        .. warning::
-            This API is experimental and may change in future versions.
-            Use with caution in production environments.
-
-        This function takes a Temporal activity function and converts it into an
-        OpenAI agent tool that can be used by the agent to execute the activity
-        during workflow execution. The tool will automatically handle the conversion
-        of inputs and outputs between the agent and the activity. Note that if you take a context,
-        mutation will not be persisted, as the activity may not be running in the same location.
-
-        Args:
-            fn: A Temporal activity function to convert to a tool.
-            For other arguments, refer to :py:mod:`workflow` :py:meth:`start_activity`
-
-        Returns:
-            An OpenAI agent tool that wraps the provided activity.
-
-        Raises:
-            ApplicationError: If the function is not properly decorated as a Temporal activity.
-
-        Example:
-            >>> @activity.defn
-            >>> def process_data(input: str) -> str:
-            ...     return f"Processed: {input}"
-            >>>
-            >>> # Create tool with custom activity options
-            >>> tool = activity_as_tool(
-            ...     process_data,
-            ...     start_to_close_timeout=timedelta(seconds=30),
-            ...     retry_policy=RetryPolicy(maximum_attempts=3),
-            ...     heartbeat_timeout=timedelta(seconds=10)
-            ... )
-            >>> # Use tool with an OpenAI agent
-        """
-        ret = activity._Definition.from_callable(fn)
-        if not ret:
-            raise ApplicationError(
-                "Bare function without tool and activity decorators is not supported",
-                "invalid_tool",
-            )
-        schema = function_schema(fn)
-
-        async def run_activity(ctx: RunContextWrapper[Any], input: str) -> Any:
-            try:
-                json_data = json.loads(input)
-            except Exception as e:
-                raise ApplicationError(
-                    f"Invalid JSON input for tool {schema.name}: {input}"
-                ) from e
-
-            # Activities don't support keyword only arguments, so we can ignore the kwargs_dict return
-            args, _ = schema.to_call_args(schema.params_pydantic_model(**json_data))
-
-            # Add the context to the arguments if it takes that
-            if schema.takes_context:
-                args = [ctx] + args
-
-            result = await temporal_workflow.execute_activity(
-                fn,
-                args=args,
-                task_queue=task_queue,
-                schedule_to_close_timeout=schedule_to_close_timeout,
-                schedule_to_start_timeout=schedule_to_start_timeout,
-                start_to_close_timeout=start_to_close_timeout,
-                heartbeat_timeout=heartbeat_timeout,
-                retry_policy=retry_policy,
-                cancellation_type=cancellation_type,
-                activity_id=activity_id,
-                versioning_intent=versioning_intent,
-                summary=summary,
-                priority=priority,
-            )
-            try:
-                return str(result)
-            except Exception as e:
-                raise ToolSerializationError(
-                    "You must return a string representation of the tool output, or something we can call str() on"
-                ) from e
-
-        return FunctionTool(
-            name=schema.name,
-            description=schema.description or "",
-            params_json_schema=schema.params_json_schema,
-            on_invoke_tool=run_activity,
-            strict_json_schema=True,
-        )
-
-    @classmethod
-    def nexus_operation_as_tool(
-        cls,
-        operation: nexusrpc.Operation[Any, Any],
-        *,
-        service: Type[Any],
-        endpoint: str,
-        schedule_to_close_timeout: Optional[timedelta] = None,
-    ) -> Tool:
-        """Convert a Nexus operation into an OpenAI agent tool.
-
-        .. warning::
-            This API is experimental and may change in future versions.
-            Use with caution in production environments.
-
-        This function takes a Nexus operation and converts it into an
-        OpenAI agent tool that can be used by the agent to execute the operation
-        during workflow execution. The tool will automatically handle the conversion
-        of inputs and outputs between the agent and the operation.
-
-        Args:
-            fn: A Nexus operation to convert into a tool.
-            service: The Nexus service class that contains the operation.
-            endpoint: The Nexus endpoint to use for the operation.
-
-        Returns:
-            An OpenAI agent tool that wraps the provided operation.
-
-        Example:
-            >>> @nexusrpc.service
-            ... class WeatherService:
-            ...     get_weather_object_nexus_operation: nexusrpc.Operation[WeatherInput, Weather]
-            >>>
-            >>> # Create tool with custom activity options
-            >>> tool = nexus_operation_as_tool(
-            ...     WeatherService.get_weather_object_nexus_operation,
-            ...     service=WeatherService,
-            ...     endpoint="weather-service",
-            ... )
-            >>> # Use tool with an OpenAI agent
-        """
-
-        def operation_callable(input):
-            raise NotImplementedError("This function definition is used as a type only")
-
-        operation_callable.__annotations__ = {
-            "input": operation.input_type,
-            "return": operation.output_type,
-        }
-        operation_callable.__name__ = operation.name
-
-        schema = function_schema(operation_callable)
-
-        async def run_operation(ctx: RunContextWrapper[Any], input: str) -> Any:
-            try:
-                json_data = json.loads(input)
-            except Exception as e:
-                raise ApplicationError(
-                    f"Invalid JSON input for tool {schema.name}: {input}"
-                ) from e
-
-            nexus_client = temporal_workflow.create_nexus_client(
-                service=service, endpoint=endpoint
-            )
-            args, _ = schema.to_call_args(schema.params_pydantic_model(**json_data))
-            assert len(args) == 1, "Nexus operations must have exactly one argument"
-            [arg] = args
-            result = await nexus_client.execute_operation(
-                operation,
-                arg,
-                schedule_to_close_timeout=schedule_to_close_timeout,
-            )
-            try:
-                return str(result)
-            except Exception as e:
-                raise ToolSerializationError(
-                    "You must return a string representation of the tool output, or something we can call str() on"
-                ) from e
-
-        return FunctionTool(
-            name=schema.name,
-            description=schema.description or "",
-            params_json_schema=schema.params_json_schema,
-            on_invoke_tool=run_operation,
-            strict_json_schema=True,
-        )