temporalio.contrib.openai_agents.workflow

module documentation
(source)

Workflow-specific primitives for working with the OpenAI Agents SDK in a workflow context

Exception ToolSerializationError Error that occurs when a tool output could not be serialized.
Function activity_as_tool Convert a single Temporal activity function to an OpenAI agent tool.
Function nexus_operation_as_tool Convert a Nexus operation into an OpenAI agent tool.
def activity_as_tool(fn: Callable, *, task_queue: str | None = None, schedule_to_close_timeout: timedelta | None = None, schedule_to_start_timeout: timedelta | None = None, start_to_close_timeout: timedelta | None = None, heartbeat_timeout: timedelta | None = None, retry_policy: RetryPolicy | None = None, cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL, activity_id: str | None = None, versioning_intent: VersioningIntent | None = None, summary: str | None = None, priority: Priority = Priority.default) -> Tool: (source)

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.

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
Parameters
fn:CallableA Temporal activity function to convert to a tool.
task_queue:str | NoneUndocumented
schedule_to_close_timeout:timedelta | NoneUndocumented
schedule_to_start_timeout:timedelta | NoneUndocumented
start_to_close_timeout:timedelta | NoneUndocumented
heartbeat_timeout:timedelta | NoneUndocumented
retry_policy:RetryPolicy | NoneUndocumented
cancellation_type:ActivityCancellationTypeUndocumented
activity_id:str | NoneUndocumented
versioning_intent:VersioningIntent | NoneUndocumented
summary:str | NoneUndocumented
priority:PriorityUndocumented
For other arguments
refer to workflow start_activity
Returns
ToolAn OpenAI agent tool that wraps the provided activity.
Raises
ApplicationErrorIf the function is not properly decorated as a Temporal activity.
def nexus_operation_as_tool(operation: nexusrpc.Operation[Any, Any], *, service: type[Any], endpoint: str, schedule_to_close_timeout: timedelta | None = None) -> Tool: (source)

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.

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
Parameters
operation:nexusrpc.Operation[Any, Any]Undocumented
service:type[Any]The Nexus service class that contains the operation.
endpoint:strThe Nexus endpoint to use for the operation.
schedule_to_close_timeout:timedelta | NoneUndocumented
fnA Nexus operation to convert into a tool.
Returns
ToolAn OpenAI agent tool that wraps the provided operation.