Merge branch 'main' into openai/trace_random

Author: tconley1428

pyproject.toml

@@ -57,6 +57,7 @@ dev = [
     "pytest-cov>=6.1.1",
     "httpx>=0.28.1",
     "pytest-pretty>=1.3.0",
+    "openai-agents[litellm] >= 0.2.3,<0.3"
 ]
 
 [tool.poe.tasks]

temporalio/contrib/openai_agents/_invoke_model_activity.py

@@ -7,13 +7,16 @@
 import json
 from dataclasses import dataclass
 from datetime import timedelta
-from typing import Any, Optional, Union, cast
+from typing import Any, Optional, Union
 
 from agents import (
     AgentOutputSchemaBase,
+    CodeInterpreterTool,
     FileSearchTool,
     FunctionTool,
     Handoff,
+    HostedMCPTool,
+    ImageGenerationTool,
     ModelProvider,
     ModelResponse,
     ModelSettings,
@@ -25,13 +28,12 @@
     UserError,
     WebSearchTool,
 )
-from agents.models.multi_provider import MultiProvider
 from openai import (
     APIStatusError,
     AsyncOpenAI,
-    AuthenticationError,
-    PermissionDeniedError,
 )
+from openai.types.responses.tool_param import Mcp
+from pydantic_core import to_json
 from typing_extensions import Required, TypedDict
 
 from temporalio import activity
@@ -41,7 +43,9 @@
 
 @dataclass
 class HandoffInput:
-    """Data conversion friendly representation of a Handoff."""
+    """Data conversion friendly representation of a Handoff. Contains only the fields which are needed by the model
+    execution to determine what to handoff to, not the actual handoff invocation, which remains in the workflow context.
+    """
 
     tool_name: str
     tool_description: str
@@ -52,15 +56,33 @@ class HandoffInput:
 
 @dataclass
 class FunctionToolInput:
-    """Data conversion friendly representation of a FunctionTool."""
+    """Data conversion friendly representation of a FunctionTool. Contains only the fields which are needed by the model
+    execution to determine what tool to call, not the actual tool invocation, which remains in the workflow context.
+    """
 
     name: str
     description: str
     params_json_schema: dict[str, Any]
     strict_json_schema: bool = True
 
 
-ToolInput = Union[FunctionToolInput, FileSearchTool, WebSearchTool]
+@dataclass
+class HostedMCPToolInput:
+    """Data conversion friendly representation of a HostedMCPTool. Contains only the fields which are needed by the model
+    execution to determine what tool to call, not the actual tool invocation, which remains in the workflow context.
+    """
+
+    tool_config: Mcp
+
+
+ToolInput = Union[
+    FunctionToolInput,
+    FileSearchTool,
+    WebSearchTool,
+    ImageGenerationTool,
+    CodeInterpreterTool,
+    HostedMCPToolInput,
+]
 
 
 @dataclass
@@ -152,22 +174,31 @@ async def empty_on_invoke_handoff(
 
         # workaround for https://github.com/pydantic/pydantic/issues/9541
         # ValidatorIterator returned
-        input_json = json.dumps(input["input"], default=str)
+        input_json = to_json(input["input"])
         input_input = json.loads(input_json)
 
         def make_tool(tool: ToolInput) -> Tool:
-            if isinstance(tool, FileSearchTool):
-                return cast(FileSearchTool, tool)
-            elif isinstance(tool, WebSearchTool):
-                return cast(WebSearchTool, tool)
+            if isinstance(
+                tool,
+                (
+                    FileSearchTool,
+                    WebSearchTool,
+                    ImageGenerationTool,
+                    CodeInterpreterTool,
+                ),
+            ):
+                return tool
+            elif isinstance(tool, HostedMCPToolInput):
+                return HostedMCPTool(
+                    tool_config=tool.tool_config,
+                )
             elif isinstance(tool, FunctionToolInput):
-                t = cast(FunctionToolInput, tool)
                 return FunctionTool(
-                    name=t.name,
-                    description=t.description,
-                    params_json_schema=t.params_json_schema,
+                    name=tool.name,
+                    description=tool.description,
+                    params_json_schema=tool.params_json_schema,
                     on_invoke_tool=empty_on_invoke_tool,
-                    strict_json_schema=t.strict_json_schema,
+                    strict_json_schema=tool.strict_json_schema,
                 )
             else:
                 raise UserError(f"Unknown tool type: {tool.name}")

temporalio/contrib/openai_agents/_openai_runner.py

@@ -7,6 +7,7 @@
     RunConfig,
     RunResult,
     RunResultStreaming,
+    SQLiteSession,
     TContext,
     Tool,
     TResponseInputItem,
@@ -51,23 +52,33 @@ async def run(
                     "Provided tool is not a tool type. If using an activity, make sure to wrap it with openai_agents.workflow.activity_as_tool."
                 )
 
+        if starting_agent.mcp_servers:
+            raise ValueError(
+                "Temporal OpenAI agent does not support on demand MCP servers."
+            )
+
         context = kwargs.get("context")
         max_turns = kwargs.get("max_turns", DEFAULT_MAX_TURNS)
         hooks = kwargs.get("hooks")
         run_config = kwargs.get("run_config")
         previous_response_id = kwargs.get("previous_response_id")
+        session = kwargs.get("session")
+
+        if isinstance(session, SQLiteSession):
+            raise ValueError("Temporal workflows don't support SQLite sessions.")
 
         if run_config is None:
             run_config = RunConfig()
 
-        if run_config.model is not None and not isinstance(run_config.model, str):
+        model_name = run_config.model or starting_agent.model
+        if model_name is not None and not isinstance(model_name, str):
             raise ValueError(
-                "Temporal workflows require a model name to be a string in the run config."
+                "Temporal workflows require a model name to be a string in the run config and/or agent."
             )
         updated_run_config = replace(
             run_config,
             model=_TemporalModelStub(
-                run_config.model,
+                model_name=model_name,
                 model_params=self.model_params,
             ),
         )
@@ -80,6 +91,7 @@ async def run(
             hooks=hooks,
             run_config=updated_run_config,
             previous_response_id=previous_response_id,
+            session=session,
         )
 
     def run_sync(

temporalio/contrib/openai_agents/_temporal_model_stub.py

@@ -8,15 +8,17 @@
 
 logger = logging.getLogger(__name__)
 
-from typing import Any, AsyncIterator, Sequence, Union, cast
+from typing import Any, AsyncIterator, Union, cast
 
 from agents import (
     AgentOutputSchema,
     AgentOutputSchemaBase,
-    ComputerTool,
+    CodeInterpreterTool,
     FileSearchTool,
     FunctionTool,
     Handoff,
+    HostedMCPTool,
+    ImageGenerationTool,
     Model,
     ModelResponse,
     ModelSettings,
@@ -33,6 +35,7 @@
     AgentOutputSchemaInput,
     FunctionToolInput,
     HandoffInput,
+    HostedMCPToolInput,
     ModelActivity,
     ModelTracingInput,
     ToolInput,
@@ -54,7 +57,7 @@ def __init__(
     async def get_response(
         self,
         system_instructions: Optional[str],
-        input: Union[str, list[TResponseInputItem], dict[str, str]],
+        input: Union[str, list[TResponseInputItem]],
         model_settings: ModelSettings,
         tools: list[Tool],
         output_schema: Optional[AgentOutputSchemaBase],
@@ -64,35 +67,19 @@ async def get_response(
         previous_response_id: Optional[str],
         prompt: Optional[ResponsePromptParam],
     ) -> ModelResponse:
-        def get_summary(
-            input: Union[str, list[TResponseInputItem], dict[str, str]],
-        ) -> str:
-            ### Activity summary shown in the UI
-            try:
-                max_size = 100
-                if isinstance(input, str):
-                    return input[:max_size]
-                elif isinstance(input, list):
-                    seq_input = cast(Sequence[Any], input)
-                    last_item = seq_input[-1]
-                    if isinstance(last_item, dict):
-                        return last_item.get("content", "")[:max_size]
-                    elif hasattr(last_item, "content"):
-                        return str(getattr(last_item, "content"))[:max_size]
-                    return str(last_item)[:max_size]
-                elif isinstance(input, dict):
-                    return input.get("content", "")[:max_size]
-            except Exception as e:
-                logger.error(f"Error getting summary: {e}")
-            return ""
-
         def make_tool_info(tool: Tool) -> ToolInput:
-            if isinstance(tool, (FileSearchTool, WebSearchTool)):
+            if isinstance(
+                tool,
+                (
+                    FileSearchTool,
+                    WebSearchTool,
+                    ImageGenerationTool,
+                    CodeInterpreterTool,
+                ),
+            ):
                 return tool
-            elif isinstance(tool, ComputerTool):
-                raise NotImplementedError(
-                    "Computer search preview is not supported in Temporal model"
-                )
+            elif isinstance(tool, HostedMCPTool):
+                return HostedMCPToolInput(tool_config=tool.tool_config)
             elif isinstance(tool, FunctionTool):
                 return FunctionToolInput(
                     name=tool.name,
@@ -101,7 +88,7 @@ def make_tool_info(tool: Tool) -> ToolInput:
                     strict_json_schema=tool.strict_json_schema,
                 )
             else:
-                raise ValueError(f"Unknown tool type: {tool.name}")
+                raise ValueError(f"Unsupported tool type: {tool.name}")
 
         tool_infos = [make_tool_info(x) for x in tools]
         handoff_infos = [
@@ -150,7 +137,7 @@ def make_tool_info(tool: Tool) -> ToolInput:
         return await workflow.execute_activity_method(
             ModelActivity.invoke_model_activity,
             activity_input,
-            summary=self.model_params.summary_override or get_summary(input),
+            summary=self.model_params.summary_override or _extract_summary(input),
             task_queue=self.model_params.task_queue,
             schedule_to_close_timeout=self.model_params.schedule_to_close_timeout,
             schedule_to_start_timeout=self.model_params.schedule_to_start_timeout,
@@ -176,3 +163,34 @@ def stream_response(
         prompt: ResponsePromptParam | None,
     ) -> AsyncIterator[TResponseStreamEvent]:
         raise NotImplementedError("Temporal model doesn't support streams yet")
+
+
+def _extract_summary(input: Union[str, list[TResponseInputItem]]) -> str:
+    ### Activity summary shown in the UI
+    try:
+        max_size = 100
+        if isinstance(input, str):
+            return input[:max_size]
+        elif isinstance(input, list):
+            # Find all message inputs, which are reasonably summarizable
+            messages: list[TResponseInputItem] = [
+                item for item in input if item.get("type", "message") == "message"
+            ]
+            if not messages:
+                return ""
+
+            content: Any = messages[-1].get("content", "")
+
+            # In the case of multiple contents, take the last one
+            if isinstance(content, list):
+                if not content:
+                    return ""
+                content = content[-1]
+
+            # Take the text field from the content if present
+            if isinstance(content, dict) and content.get("text") is not None:
+                content = content.get("text")
+            return str(content)[:max_size]
+    except Exception as e:
+        logger.error(f"Error getting summary: {e}")
+    return ""

temporalio/contrib/openai_agents/_temporal_openai_agents.py

@@ -34,7 +34,13 @@
 from temporalio.contrib.openai_agents._trace_interceptor import (
     OpenAIAgentsTracingInterceptor,
 )
-from temporalio.contrib.pydantic import pydantic_data_converter
+from temporalio.contrib.pydantic import (
+    PydanticPayloadConverter,
+    ToJsonOptions,
+)
+from temporalio.converter import (
+    DataConverter,
+)
 from temporalio.worker import Worker, WorkerConfig
 
 
@@ -137,6 +143,11 @@ def stream_response(
         raise NotImplementedError()
 
 
+class _OpenAIPayloadConverter(PydanticPayloadConverter):
+    def __init__(self) -> None:
+        super().__init__(ToJsonOptions(exclude_unset=True))
+
+
 class OpenAIAgentsPlugin(temporalio.client.Plugin, temporalio.worker.Plugin):
     """Temporal plugin for integrating OpenAI agents with Temporal workflows.
 
@@ -232,7 +243,9 @@ def configure_client(self, config: ClientConfig) -> ClientConfig:
         Returns:
             The modified client configuration.
         """
-        config["data_converter"] = pydantic_data_converter
+        config["data_converter"] = DataConverter(
+            payload_converter_class=_OpenAIPayloadConverter
+        )
         return super().configure_client(config)
 
     def configure_worker(self, config: WorkerConfig) -> WorkerConfig:

temporalio/contrib/openai_agents/workflow.py

@@ -134,7 +134,7 @@ async def run_activity(ctx: RunContextWrapper[Any], input: str) -> Any:
             cancellation_type=cancellation_type,
             activity_id=activity_id,
             versioning_intent=versioning_intent,
-            summary=summary,
+            summary=summary or schema.description,
             priority=priority,
         )
         try: