Before running these examples, be sure to review the [prerequisites and background on the integration](../README.md).

Basic agent that only responds in haikus:

Agent with access to external tools (simulated weather API):

Demonstrates agent lifecycle events and handoffs between agents:

uv run openai_agents/basic/run_agent_lifecycle_workflow.py

Shows detailed usage tracking with RunHooks (requests, tokens, etc.):

uv run openai_agents/basic/run_lifecycle_workflow.py

Agent with dynamic instruction generation based on context (haiku/pirate/robot):

uv run openai_agents/basic/run_dynamic_system_prompt_workflow.py

Demonstrates different JSON schema validation approaches:

uv run openai_agents/basic/run_non_strict_output_workflow.py

Note: `CustomOutputSchema` is not supported by the Temporal OpenAI Agents SDK integration and is omitted in this example.

Process local image files with AI vision:

uv run openai_agents/basic/run_local_image_workflow.py

Process remote image URLs with AI vision:

uv run openai_agents/basic/run_remote_image_workflow.py

Demonstrates conversation continuity using response IDs:

uv run openai_agents/basic/run_previous_response_id_workflow.py

The following examples from the [reference repository](https://github.com/openai/openai-agents-python/tree/main/examples/basic) are not included in this Temporal adaptation:

- **Session** - Stores state in local SQLite database, not appropriate for distributed workflows

- **Stream Items/Stream Text** - Streaming is not supported in Temporal OpenAI Agents SDK integration

import base64

from temporalio import activity

async def read_image_as_base64(image_path: str) -> str:

"""

with open(image_path, "rb") as image_file:

encoded_string = base64.b64encode(image_file.read()).decode("utf-8")

return encoded_string

import random

from temporalio import activity

async def random_number(max_value: int) -> int:

"""Generate a random number up to the provided maximum."""

return random.randint(0, max_value)

async def multiply_by_two(x: int) -> int:

"""Simple multiplication by two."""

return x * 2

import asyncio

from temporalio.client import Client

from openai_agents.basic.workflows.agent_lifecycle_workflow import (

AgentLifecycleWorkflow,

)

async def main() -> None:

client = await Client.connect("localhost:7233")

user_input = input("Enter a max number: ")

max_number = int(user_input)

result = await client.execute_workflow(

AgentLifecycleWorkflow.run,

max_number,

id="agent-lifecycle-workflow",

task_queue="openai-agents-basic-task-queue",

)

print(f"Final result: {result}")

print("Done!")

if __name__ == "__main__":

asyncio.run(main())

import asyncio

from temporalio.client import Client

from temporalio.contrib.openai_agents import OpenAIAgentsPlugin

from openai_agents.basic.workflows.dynamic_system_prompt_workflow import (

DynamicSystemPromptWorkflow,

)

async def main():

client = await Client.connect(

"localhost:7233",

plugins=[

OpenAIAgentsPlugin(),

],

)

user_message = "Tell me a joke."

result = await client.execute_workflow(

DynamicSystemPromptWorkflow.run,

user_message,

id="dynamic-prompt-workflow",

task_queue="openai-agents-basic-task-queue",

)

print(result)

print()

# Run with specific style

result = await client.execute_workflow(

DynamicSystemPromptWorkflow.run,

args=[user_message, "pirate"],

id="dynamic-prompt-pirate-workflow",

task_queue="openai-agents-basic-task-queue",

)

print(result)

if __name__ == "__main__":

asyncio.run(main())

task_queue="openai-agents-basic-task-queue",

import asyncio

from temporalio.client import Client

from temporalio.contrib.openai_agents import OpenAIAgentsPlugin

from openai_agents.basic.workflows.lifecycle_workflow import LifecycleWorkflow

async def main():

client = await Client.connect(

"localhost:7233",

plugins=[

OpenAIAgentsPlugin(),

],

)

user_input = input("Enter a max number: ")

max_number = int(user_input)

result = await client.execute_workflow(

LifecycleWorkflow.run,

max_number,

id="lifecycle-workflow",

task_queue="openai-agents-basic-task-queue",

)

print(f"Final result: {result}")

if __name__ == "__main__":

asyncio.run(main())

import asyncio

import os

from temporalio.client import Client

from temporalio.contrib.openai_agents import OpenAIAgentsPlugin

from openai_agents.basic.workflows.local_image_workflow import LocalImageWorkflow

async def main():

client = await Client.connect(

"localhost:7233",

plugins=[

OpenAIAgentsPlugin(),

],

)

# Use the media file from the original example

image_path = os.path.join(os.path.dirname(__file__), "media/image_bison.jpg")

result = await client.execute_workflow(

LocalImageWorkflow.run,

args=[image_path, "What do you see in this image?"],

id="local-image-workflow",

task_queue="openai-agents-basic-task-queue",

)

print(f"Agent response: {result}")

if __name__ == "__main__":

asyncio.run(main())

import asyncio

from temporalio.client import Client

from temporalio.contrib.openai_agents import OpenAIAgentsPlugin

from openai_agents.basic.workflows.non_strict_output_workflow import (

NonStrictOutputWorkflow,

)

async def main():

client = await Client.connect(

"localhost:7233",

plugins=[

OpenAIAgentsPlugin(),

],

)

input_message = "Tell me 3 short jokes."

result = await client.execute_workflow(

NonStrictOutputWorkflow.run,

input_message,

id="non-strict-output-workflow",

task_queue="openai-agents-basic-task-queue",

)

print("=== Non-Strict Output Type Results ===")

for key, value in result.items():

print(f"{key}: {value}")

print()

if __name__ == "__main__":

asyncio.run(main())