temporalio.contrib.openai_agents.StatefulMCPServerProvider

class documentation

class StatefulMCPServerProvider: (source)

Constructor: StatefulMCPServerProvider(name, server_factory)

A stateful MCP server implementation for Temporal workflows.

This class wraps a function to create MCP servers to maintain a persistent connection throughout the workflow execution. It creates a dedicated worker that stays connected to the MCP server and processes operations on a dedicated task queue.

This approach will allow the MCPServer to maintain state across calls if needed, but the caller will have to handle cases where the dedicated worker fails, as Temporal is unable to seamlessly recreate any lost state in that case. It is discouraged to use this approach unless necessary.

Handling dedicated worker failure will entail catching ApplicationError with type "DedicatedWorkerFailure". Depending on the usage pattern, the caller will then have to either restart from the point at which the Stateful server was needed or handle continuing from that loss of state in some other way.

Method	`__init__`	Initialize the stateful temporal MCP server.
Property	`name`	Get the server name.
Method	`_get_activities`	Undocumented
Instance Variable	`_connect_handle`	Undocumented
Instance Variable	`_name`	Undocumented
Instance Variable	`_server_factory`	Undocumented
Instance Variable	`_servers`	Undocumented

def __init__(self, name: str, server_factory: Callable[[(Any | None)], MCPServer]): (source) ¶

Initialize the stateful temporal MCP server.

Parameters
name:`str`	The name of the MCP server.
server_factory:`Callable[[(Any \| None)], MCPServer]`	A function which will produce MCPServer instances. It should return a new server each time so that state is not shared between workflow runs. It receives an optional `factory_argument` from the workflow.

@property

name: str = (source) ¶

Get the server name.

def _get_activities(self) -> Sequence[Callable]: (source) ¶

Undocumented

_connect_handle: ActivityHandle | None = (source) ¶

Undocumented

_name = (source) ¶

Undocumented

_server_factory = (source) ¶

Undocumented

_servers: dict[str, MCPServer] = (source) ¶

Undocumented