temporalio.contrib.openai_agents.testing.AgentEnvironment

class documentation

class AgentEnvironment: (source)

Constructor: AgentEnvironment(model_params, model_provider, model, mcp_server_providers, ...)

Testing environment for OpenAI agents with Temporal integration.

This async context manager provides a convenient way to set up testing environments for OpenAI agents with mocked model calls and Temporal integration.

Warning

This API is experimental and may change in the future.

Example

>>> from temporalio.contrib.openai_agents.testing import AgentEnvironment, TestModelProvider, ResponseBuilders
>>> from temporalio.client import Client
>>>
>>> # Create a mock model that returns predefined responses
>>> mock_model = TestModel.returning_responses([
...     ResponseBuilders.output_message("Hello, world!"),
...     ResponseBuilders.output_message("How can I help you?")
... ])
>>>
>>> async with AgentEnvironment(model=mock_model) as env:
...     client = env.applied_on_client(client)
...     # Use client for testing workflows with mocked model calls

Async Method	`__aenter__`	Enter the async context manager.
Async Method	`__aexit__`	Exit the async context manager.
Method	`__init__`	Initialize the AgentEnvironment.
Method	`applied_on_client`	Apply the agent environment's plugin to a client and return a new client instance.
Class Variable	`__test__`	Undocumented
Property	`openai_agents_plugin`	Get the underlying OpenAI agents plugin.
Instance Variable	`_mcp_server_providers`	Undocumented
Instance Variable	`_model_params`	Undocumented
Instance Variable	`_model_provider`	Undocumented
Instance Variable	`_plugin`	Undocumented
Instance Variable	`_register_activities`	Undocumented

async def __aenter__(self) -> AgentEnvironment: (source) ¶

Enter the async context manager.

async def __aexit__(self, *args): (source) ¶

Exit the async context manager.

def __init__(self, model_params: ModelActivityParameters | None = None, model_provider: ModelProvider | None = None, model: Model | None = None, mcp_server_providers: Sequence[StatelessMCPServerProvider | StatefulMCPServerProvider] = (), register_activities: bool = True): (source) ¶

Initialize the AgentEnvironment.

Warning

This API is experimental and may change in the future.

Parameters
model_params:`ModelActivityParameters \| None`	Configuration parameters for Temporal activity execution of model calls. If None, default parameters will be used.
model_provider:`ModelProvider \| None`	Optional model provider for custom model implementations. Only one of model_provider or model should be provided. If both are provided, model_provider will be used.
model:`Model \| None`	Optional model for custom model implementations. Use TestModel for mocking model responses. Equivalent to model_provider=TestModelProvider(model). Only one of model_provider or model should be provided. If both are provided, model_provider will be used.
mcp_server_providers:`Sequence[StatelessMCPServerProvider \| StatefulMCPServerProvider]`	Sequence of MCP servers to automatically register with the worker.
register_activities:`bool`	Whether to register activities during worker execution.