temporalio.contrib.openai_agents.StatefulMCPServerProvider

class documentation

class StatefulMCPServerProvider: (source)

Constructor: StatefulMCPServerProvider(server_factory)

View In Hierarchy

A stateful MCP server implementation for Temporal workflows.

This class wraps an 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, server_factory: Callable[[], MCPServer]): (source)

Initialize the stateful temporal MCP server.

Parameters
server_factory:Callable[[], 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
@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