temporalio.contrib.openai_agents.temporal_tools

module documentation
(source)

Support for using Temporal activities as OpenAI agents tools.

Function activity_as_tool Convert a single Temporal activity function to an OpenAI agent tool.
def activity_as_tool(fn: Callable, **kwargs) -> 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.

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.
**kwargsAdditional arguments to pass to workflow.execute_activity. These arguments configure how the activity is executed. Common options include: - start_to_close_timeout: Maximum time for the activity to complete - schedule_to_close_timeout: Maximum time from scheduling to completion - schedule_to_start_timeout: Maximum time from scheduling to starting - heartbeat_timeout: Maximum time between heartbeats - retry_policy: Policy for retrying failed activities - task_queue: Specific task queue to use for this activity - cancellation_type: How the activity handles cancellation - workflow_id_reuse_policy: Policy for workflow ID reuse
Returns
ToolAn OpenAI agent tool that wraps the provided activity.
Raises
ApplicationErrorIf the function is not properly decorated as a Temporal activity.