diff --git a/python/openai_agents/README.md b/python/openai_agents/README.md index 25c774d5..092ca0ff 100644 --- a/python/openai_agents/README.md +++ b/python/openai_agents/README.md @@ -447,8 +447,8 @@ worker-side OpenAI `MCPServer`. Configure workflow-facing behavior such as openai-agents 0.22.3, tool input and output guardrails on `temporal_mcp_server(...)`, where the OpenAI agent can use it. On the worker-side server, a static `tool_filter` still filters the listed tools, a -callable `tool_filter` and a `require_approval` policy are rejected as described -below, and the other settings are ignored. Custom `MCPServer` method +callable `tool_filter`, a `require_approval` policy and a `tool_meta_resolver` +are rejected as described below, and the other settings are ignored. Custom `MCPServer` method implementations still execute worker-side. A callable `tool_filter` on the worker-side server is rejected, because the run @@ -463,13 +463,22 @@ Agents SDK normalizes to no approval is allowed there: `"never"`, `False`, or a mapping or tool list with only `"never"` entries. A callable policy is always rejected, because it cannot be evaluated on the worker. +A `tool_meta_resolver` on the worker-side server is rejected too, because the +run context it receives exists only in the workflow. Pass it to +`temporal_mcp_server(tool_meta_resolver=...)` instead, where it runs in the +workflow and must be deterministic. Its result is recorded in workflow history, +so it must not hold secrets. + These checks run in every MCP Activity. A run that already holds its cached tool list, the default, first meets them when it calls a tool: the agent gets the error as the tool's output, and the run can complete without the tool. Let open runs finish before you upgrade or add such a setting to the factory's server. +A resolver there never had an effect, so removing it from the factory's server +in the same deploy is enough. Moving the policy into `temporal_mcp_server(require_approval=...)` is also a workflow change: an open run that already called one of that server's tools -fails replay with a nondeterminism error. +fails replay with a nondeterminism error. Moving a resolver there is a workflow +change too, so guard it with `workflow.patched()` or let open runs finish first. Every MCP operation is a Temporal Activity. The workflow-side tool list is cached by default; pass `cache_tools_list=False` to refresh it on every Agents @@ -540,10 +549,17 @@ The legacy stateful path retains its dedicated per-workflow worker and persistent-session behavior. To require tool approval on these servers, pass `require_approval` to `stateless_mcp_server()` or `stateful_mcp_server()`; as with `temporal_mcp_server()`, a policy on the factory's server is rejected -unless it never asks for approval. So is a callable `tool_filter` there, in -every MCP operation, direct `call_tool()` and prompt calls included; a static -filter still works. Remove such a filter from the factory in the same deploy, -or let open runs finish first. +unless it never asks for approval. So are a callable `tool_filter` and a +`tool_meta_resolver` there, in every MCP operation, direct `call_tool()` and +prompt calls included; a static filter still works. On these routes, set a +resolver on the server that `stateless_mcp_server()` or `stateful_mcp_server()` +gives the workflow (`server.tool_meta_resolver = ...`); it then gets +`-stateless` or `-stateful` as `server_name`. As with +`temporal_mcp_server()`, it runs in the workflow and must be deterministic, its +result is recorded in workflow history, and adding it to a workflow with open +runs needs `workflow.patched()` or those runs to finish first. Remove such a +filter or resolver from the factory in the same deploy, or let open runs finish +first. ### Hosted MCP Tool diff --git a/python/openai_agents/src/temporalio/openai_agents/_mcp.py b/python/openai_agents/src/temporalio/openai_agents/_mcp.py index fac7c391..17ab3299 100644 --- a/python/openai_agents/src/temporalio/openai_agents/_mcp.py +++ b/python/openai_agents/src/temporalio/openai_agents/_mcp.py @@ -29,8 +29,10 @@ from temporalio.openai_agents._mcp_server_checks import ( _has_dynamic_tool_filter, _has_worker_side_require_approval, + _has_worker_side_tool_meta_resolver, _reject_dynamic_tool_filter, _reject_worker_side_require_approval, + _reject_worker_side_tool_meta_resolver, ) from temporalio.worker import PollerBehaviorSimpleMaximum, Worker from temporalio.workflow import ActivityConfig, ActivityHandle @@ -43,6 +45,15 @@ ) +def _legacy_tool_meta_resolver_remedy(route: str) -> str: + """What the deprecated routes' tool_meta_resolver error tells the user to do.""" + return ( + f"set tool_meta_resolver on the server {route}() gives the workflow, or " + "register the factory with OpenAIAgentsPlugin(mcp_servers=...) and pass " + "it to temporal_mcp_server(tool_meta_resolver=...)" + ) + + @dataclasses.dataclass class _StatelessListToolsArguments: factory_argument: Any | None @@ -173,9 +184,9 @@ def __init__( name: The name of the MCP server. server_factory: 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 may accept a single positional parameter, which - receives a ``factory_argument`` from the workflow. A callable ``tool_filter`` or a ``require_approval`` - policy that can ask for approval on the returned server is rejected in every MCP operation; a static - filter is applied. + receives a ``factory_argument`` from the workflow. A callable ``tool_filter``, a ``require_approval`` + policy that can ask for approval, or a ``tool_meta_resolver`` on the returned server is rejected in + every MCP operation; a static filter is applied. """ warnings.warn( "StatelessMCPServerProvider is deprecated; use " @@ -207,6 +218,11 @@ def _create_server(self, factory_argument: Any | None) -> MCPServer: _reject_worker_side_require_approval( self._server_name, server, "stateless_mcp_server" ) + _reject_worker_side_tool_meta_resolver( + self._server_name, + server, + _legacy_tool_meta_resolver_remedy("stateless_mcp_server"), + ) return server @property @@ -481,8 +497,9 @@ def __init__( name: The name of the MCP server. server_factory: 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. A callable ``tool_filter`` or a ``require_approval`` policy that can ask for approval on the - returned server is rejected in every MCP operation; a static filter is applied. + workflow. A callable ``tool_filter``, a ``require_approval`` policy that can ask for approval, or a + ``tool_meta_resolver`` on the returned server is rejected in every MCP operation; a static filter is + applied. """ warnings.warn( "StatefulMCPServerProvider is deprecated; use " @@ -515,6 +532,11 @@ def _server() -> MCPServer: _reject_worker_side_require_approval( self._server_name, server, "stateful_mcp_server" ) + _reject_worker_side_tool_meta_resolver( + self._server_name, + server, + _legacy_tool_meta_resolver_remedy("stateful_mcp_server"), + ) return server @activity.defn(name=self.name + "-list-tools") @@ -556,13 +578,17 @@ async def connect( "Cannot connect to an already running server. Use a distinct name if running multiple servers in one workflow." ) server = self._server_factory(args.factory_argument if args else None) - # A server with a callable tool filter or a worker-side approval - # policy is never connected. The dedicated worker still runs, and + # A server with a callable tool filter, a worker-side approval + # policy or a tool_meta_resolver is never connected. The dedicated worker still runs, and # each of its operations fails with the same error, so the workflow # sees that error at its first MCP call whatever its # schedule-to-start timeout. has_filter = _has_dynamic_tool_filter(server) - rejected = has_filter or _has_worker_side_require_approval(server) + rejected = ( + has_filter + or _has_worker_side_require_approval(server) + or _has_worker_side_tool_meta_resolver(server) + ) self._servers[server_id] = server try: try: diff --git a/python/openai_agents/src/temporalio/openai_agents/_mcp_backend.py b/python/openai_agents/src/temporalio/openai_agents/_mcp_backend.py index b864d7bb..11131990 100644 --- a/python/openai_agents/src/temporalio/openai_agents/_mcp_backend.py +++ b/python/openai_agents/src/temporalio/openai_agents/_mcp_backend.py @@ -29,6 +29,7 @@ from temporalio.openai_agents._mcp_server_checks import ( _reject_dynamic_tool_filter, _reject_worker_side_require_approval, + _reject_worker_side_tool_meta_resolver, ) _MCPServerFactory = Callable[[], MCPServer] | Callable[[Any], MCPServer] @@ -162,6 +163,7 @@ def create(argument: Any = _NOT_SUPPLIED) -> _OpenAIMCPServerBackend: server = cast(MCPServer, invoke(argument)) _reject_dynamic_tool_filter(name, server) _reject_worker_side_require_approval(name, server, "temporal_mcp_server") + _reject_worker_side_tool_meta_resolver(name, server) return _OpenAIMCPServerBackend(server) return create diff --git a/python/openai_agents/src/temporalio/openai_agents/_mcp_server_checks.py b/python/openai_agents/src/temporalio/openai_agents/_mcp_server_checks.py index a15815df..4c52162a 100644 --- a/python/openai_agents/src/temporalio/openai_agents/_mcp_server_checks.py +++ b/python/openai_agents/src/temporalio/openai_agents/_mcp_server_checks.py @@ -71,6 +71,32 @@ def _has_worker_side_require_approval(server: MCPServer) -> bool: return callable(policy) or bool(policy) or overridden +def _has_worker_side_tool_meta_resolver(server: MCPServer) -> bool: + """Return whether the server sets a ``tool_meta_resolver``.""" + return getattr(server, "tool_meta_resolver", None) is not None + + +def _reject_worker_side_tool_meta_resolver( + name: str, + server: MCPServer, + remedy: str = "pass it to temporal_mcp_server(tool_meta_resolver=...)", +) -> None: + """Raise when the factory's server has a ``tool_meta_resolver``; see the message. + + The resolver needs the run context, which exists only in the workflow, so the + worker cannot run it, and the agent never reads the factory's copy. ``remedy`` + says where to set it instead. + """ + if _has_worker_side_tool_meta_resolver(server): + raise ApplicationError( + f"MCP server {name!r} sets tool_meta_resolver on its worker-side " + "MCPServer, which cannot run there because the run context it " + f"receives exists only in the workflow; {remedy}, and remove it from " + "the factory's server", + non_retryable=True, + ) + + def _reject_worker_side_require_approval( name: str, server: MCPServer, route: str ) -> None: diff --git a/python/openai_agents/src/temporalio/openai_agents/_temporal_openai_agents.py b/python/openai_agents/src/temporalio/openai_agents/_temporal_openai_agents.py index b8803a58..63441cbf 100644 --- a/python/openai_agents/src/temporalio/openai_agents/_temporal_openai_agents.py +++ b/python/openai_agents/src/temporalio/openai_agents/_temporal_openai_agents.py @@ -372,9 +372,9 @@ def __init__( mcp_server_providers: Legacy MCP server providers to register with the worker. Deprecated; use ``mcp_servers`` for new MCP v2 integrations. A provider's server with a callable - ``tool_filter`` or a ``require_approval`` policy that can ask for - approval is rejected in every MCP operation; a static filter is - applied. + ``tool_filter``, a ``require_approval`` policy that can ask for + approval, or a ``tool_meta_resolver`` is rejected in every MCP + operation; a static filter is applied. mcp_servers: Named OpenAI MCP server factories to register on the worker. Each factory may declare at most one positional parameter, which receives the workflow's ``factory_argument`` (``None`` when the @@ -382,9 +382,9 @@ def __init__( parameterless factory's modern connection is cached between Activities. A factory called with a workflow ``factory_argument`` always creates a fresh server for that Activity. A factory's server - with a callable ``tool_filter`` or a ``require_approval`` policy - that can ask for approval is rejected; pass those to - ``workflow.temporal_mcp_server()``. + with a callable ``tool_filter``, a ``require_approval`` policy + that can ask for approval, or a ``tool_meta_resolver`` is + rejected; pass those to ``workflow.temporal_mcp_server()``. mcp_connection_idle_timeout: How long an idle modern MCP connection remains cached. Defaults to five minutes. None disables idle eviction, and zero disables reuse. diff --git a/python/openai_agents/src/temporalio/openai_agents/workflow.py b/python/openai_agents/src/temporalio/openai_agents/workflow.py index dfc4d4ac..f88ec755 100644 --- a/python/openai_agents/src/temporalio/openai_agents/workflow.py +++ b/python/openai_agents/src/temporalio/openai_agents/workflow.py @@ -508,8 +508,9 @@ def temporal_mcp_server( of using the worker connection cache. server_options: OpenAI Agents ``MCPServer`` options such as ``require_approval`` and ``failure_error_function``. Set them here: - on the factory's server, a callable ``tool_filter`` and a - ``require_approval`` policy that can ask for approval are rejected. + on the factory's server, a callable ``tool_filter``, a + ``require_approval`` policy that can ask for approval and a + ``tool_meta_resolver`` are rejected. """ try: from temporalio.mcp import TemporalMCPClient diff --git a/python/openai_agents/tests/test_mcp_v2.py b/python/openai_agents/tests/test_mcp_v2.py index 00f738e7..90d224ad 100644 --- a/python/openai_agents/tests/test_mcp_v2.py +++ b/python/openai_agents/tests/test_mcp_v2.py @@ -75,6 +75,27 @@ async def run(self) -> str: return result.final_output +@workflow.defn +class MCPToolMetaWorkflow: + @workflow.run + async def run(self) -> str: + server = openai_agents.workflow.temporal_mcp_server( + "hello", + tool_meta_resolver=lambda context: { + "tenant": "t-7", + "tool": context.tool_name, + "server": context.server_name, + }, + ) + agent = Agent[str]( + name="MCP workflow", + instructions="Use the tools.", + mcp_servers=[server], + ) + result = await Runner.run(agent, "Say hello to Tom and Tim.") + return result.final_output + + def model() -> TestModel: return TestModel.returning_responses( [ @@ -214,6 +235,39 @@ def factory(argument: Any) -> AgentsMCPServer: assert arguments == [{"tenant": "acme"}] * 3 +async def test_temporal_mcp_server_sends_tool_meta(client: Client) -> None: + metas: list[Any] = [] + mcp_server = server() + + class RecordingServer(InProcessMCPServer): + async def call_tool( + self, + tool_name: str, + arguments: dict[str, Any] | None, + meta: dict[str, Any] | None = None, + ) -> CallToolResult: + metas.append(meta) + return await super().call_tool(tool_name, arguments, meta) + + async with AgentEnvironment( + model=model(), + model_params=ModelActivityParameters( + start_to_close_timeout=timedelta(seconds=30) + ), + mcp_servers={"hello": lambda: RecordingServer(mcp_server)}, + ) as environment: + plugged_client = environment.applied_on_client(client) + async with new_worker(plugged_client, MCPToolMetaWorkflow) as worker: + result = await plugged_client.execute_workflow( + MCPToolMetaWorkflow.run, + id=f"mcp-v2-meta-{uuid4()}", + task_queue=worker.task_queue, + ) + + assert result == "Hi Tom and Tim!" + assert metas == [{"tenant": "t-7", "tool": "say_hello", "server": "hello"}] * 2 + + def create_backend(factory: Callable[[], AgentsMCPServer]) -> _OpenAIMCPServerBackend: return cast( Callable[[], _OpenAIMCPServerBackend], @@ -233,6 +287,18 @@ def factory() -> AgentsMCPServer: assert "temporal_mcp_server()" in str(err.value) +def test_worker_side_tool_meta_resolver_is_rejected() -> None: + def factory() -> AgentsMCPServer: + value = InProcessMCPServer(server()) + value.tool_meta_resolver = lambda context: {"tenant": "t-1"} # type: ignore[attr-defined] + return value + + with pytest.raises(ApplicationError) as err: + create_backend(factory) + assert err.value.non_retryable + assert "temporal_mcp_server(tool_meta_resolver=...)" in str(err.value) + + def test_static_worker_side_tool_filter_is_allowed() -> None: def factory() -> AgentsMCPServer: value = InProcessMCPServer(server()) diff --git a/python/openai_agents/tests/test_openai.py b/python/openai_agents/tests/test_openai.py index bc25aa93..876edee0 100644 --- a/python/openai_agents/tests/test_openai.py +++ b/python/openai_agents/tests/test_openai.py @@ -3080,7 +3080,8 @@ def factory(args: Any | None) -> MCPServer: @pytest.mark.parametrize( - "setting", ["require_approval", "instance_evaluator", "tool_filter"] + "setting", + ["require_approval", "instance_evaluator", "tool_filter", "tool_meta_resolver"], ) @pytest.mark.parametrize("stateful", [True, False]) async def test_mcp_server_factory_settings_are_rejected( @@ -3094,6 +3095,9 @@ async def test_mcp_server_factory_settings_are_rejected( tracking_server._get_needs_approval_for_tool = ( # type: ignore[method-assign] lambda tool, agent: True # type: ignore[reportUnknownLambdaType] ) + elif setting == "tool_meta_resolver": + tracking_server = get_tracking_server("HelloServer") + tracking_server.tool_meta_resolver = lambda context: {"tenant": "t-1"} # type: ignore[attr-defined] else: tracking_server = get_tracking_server("HelloServer") tracking_server.tool_filter = lambda context, tool: True # type: ignore[attr-defined] @@ -3138,7 +3142,11 @@ async def test_mcp_server_factory_settings_are_rejected( assert isinstance(err.value.cause.cause, ApplicationError) assert err.value.cause.cause.non_retryable message = err.value.cause.cause.message - if setting != "tool_filter": + if setting == "tool_meta_resolver": + route = "stateful_mcp_server" if stateful else "stateless_mcp_server" + assert "'HelloServer' sets tool_meta_resolver" in message + assert f"on the server {route}() gives the workflow" in message + elif setting != "tool_filter": route = "stateful_mcp_server" if stateful else "stateless_mcp_server" assert "'HelloServer' sets require_approval" in message assert f"{route}(require_approval=...)" in message diff --git a/typescript/openai-agents/README.md b/typescript/openai-agents/README.md index 8402c90b..377e63ac 100644 --- a/typescript/openai-agents/README.md +++ b/typescript/openai-agents/README.md @@ -533,6 +533,28 @@ Because the handle caches the full list, the agent SDK's `invalidateServerToolsC Older versions also let the agent SDK cache the filtered list of a handle with a `toolFilter` property assigned, per server and agent name, so a later run with another context got the first run's tools. This version filters every listing for its own run. A run that started on an older version, whose filter depends on the run context, and that called a tool the filter now removes fails replay on this version. Let such runs finish before you upgrade. +### MCP request metadata + +A `toolMetaResolver` builds each tool call's MCP `_meta` from the run context, which exists only in the Workflow. Pass it as the `toolMetaResolver` option of `statelessMcpServer()` or `statefulMcpServer()`; it runs in Workflow code and must be deterministic. The call-tool Activity passes its result to the server's `callTool`. The Worker can't run a `toolMetaResolver` on the factory's server, so it refuses one: every list-tools and call-tool Activity for that server fails with a non-retryable `ApplicationFailure` of type `MCPToolMetaResolverUnsupported` (exported as `TOOL_META_RESOLVER_UNSUPPORTED_TYPE`). + +```ts +const server = statelessMcpServer('tickets', { + toolMetaResolver: ({ runContext }) => ({ tenant: (runContext.context as { tenant: string }).tenant }), +}); +``` + +A resolver also runs for tool calls after a run resumes from a `RunState`, for example after a tool approval, and then sees the context rebuilt from JSON: plain objects, without class methods. If it throws there, the model gets a tool error and the server isn't called. + +A new run with a `toolMetaResolver` on the factory's server fails at its first list-tools Activity. A `statelessMcpServer()` run that listed tools on an older Worker instead gets a tool error, which the model reads, for each call to that server, and the run keeps going. So remove the `toolMetaResolver` from the factory's server before you upgrade Workers. + +The `_meta` values are recorded in the call-tool Activity input in Workflow History, so don't put secrets in them. + +A call-tool Activity that runs on a Worker with an older plugin version drops the `_meta`, and nothing fails. Upgrade every Worker that runs the call-tool Activities, including Workers on a `taskQueue` you pass to `statelessMcpServer()`, before Workflows rely on `_meta`, or use Worker Versioning. + +Plugin versions before this option ignore it and don't run its resolver. An older version can replay a History only if it produces the same Workflow commands in the same order, and a resolver passed as this option can change them even when it is synchronous and never fails: the agent awaits it, which can reorder the Activities of parallel tool calls, and its changes to the run context can change later commands. An older version fails such a History with a nondeterminism error (`TMPRL1100`). Before you roll Workers back to an older plugin version, let the Workflows that use this option finish. Workflows with no resolver, or with a resolver set on the handle itself (`server.toolMetaResolver = ...`), run the same resolver code on older versions, so this option doesn't change their commands. On a `statefulMcpServer()` handle, a resolver set this way gets the handle's internal name, `-stateful`, as `serverName`; the option gets ``. + +Adding a resolver to a Workflow that has open runs, including moving one off the factory's server, can fail their replay for the same reasons, so guard the change with `patched()` or let those runs finish first. + ## Sandbox agents > **Experimental** — sandbox support is subject to change prior to General Availability. @@ -730,9 +752,9 @@ Most applications use two import paths: `@temporalio/openai-agents` in Worker an | `@temporalio/openai-agents/otel` | Worker or Client | Replay-safe OpenTelemetry setup | | `@temporalio/openai-agents/workflow-interceptor` | Worker bundling | Manual `workflowInterceptorModules` wiring without a plugin | -`@temporalio/openai-agents` exports `OpenAIAgentsPlugin`, MCP provider classes, `SandboxClientProvider`, model option types, hosted tool credential types, and `DEDICATED_WORKER_FAILURE_TYPE`. `OpenAIAgentsTraceClientInterceptor` is also public for Clients that need manual interceptor wiring instead of plugin registration. +`@temporalio/openai-agents` exports `OpenAIAgentsPlugin`, MCP provider classes, `SandboxClientProvider`, model option types, hosted tool credential types, `DEDICATED_WORKER_FAILURE_TYPE`, `TOOL_FILTER_UNSUPPORTED_TYPE` and `TOOL_META_RESOLVER_UNSUPPORTED_TYPE`. `OpenAIAgentsTraceClientInterceptor` is also public for Clients that need manual interceptor wiring instead of plugin registration. -`@temporalio/openai-agents/workflow` exports Workflow-safe APIs: `TemporalOpenAIRunner`, `WorkflowSafeMemorySession`, `activityAsTool`, `nexusOperationAsTool`, `agentAsTool`, `statelessMcpServer`, `statefulMcpServer`, `temporalSandboxClient`, `workerEnvValue`, related option/definition types, and `DEDICATED_WORKER_FAILURE_TYPE`. +`@temporalio/openai-agents/workflow` exports Workflow-safe APIs: `TemporalOpenAIRunner`, `WorkflowSafeMemorySession`, `activityAsTool`, `nexusOperationAsTool`, `agentAsTool`, `statelessMcpServer`, `statefulMcpServer`, `temporalSandboxClient`, `workerEnvValue`, related option/definition types, `DEDICATED_WORKER_FAILURE_TYPE`, `TOOL_FILTER_UNSUPPORTED_TYPE` and `TOOL_META_RESOLVER_UNSUPPORTED_TYPE`. `@temporalio/openai-agents/otel` exports `createTracerProvider`, `TemporalIdGenerator`, `markReplaySafeTracerProvider`, `isReplaySafeTracerProvider`, and `TemporalOpenAIAgentsTracerProviderOptions`. Install optional peer dependency `@opentelemetry/sdk-trace-base` only when you use this OTel integration. diff --git a/typescript/openai-agents/src/__tests__/mcp-provider.test.ts b/typescript/openai-agents/src/__tests__/mcp-provider.test.ts index 57234285..0c9e6e4a 100644 --- a/typescript/openai-agents/src/__tests__/mcp-provider.test.ts +++ b/typescript/openai-agents/src/__tests__/mcp-provider.test.ts @@ -1,6 +1,7 @@ import test from 'ava'; import type { MCPServer } from '@openai/agents-core'; import { ApplicationFailure } from '@temporalio/common'; +import { TOOL_META_RESOLVER_UNSUPPORTED_TYPE } from '..'; import { TOOL_FILTER_UNSUPPORTED_TYPE } from '../common/mcp-types'; import { StatelessMCPServerProvider, filterListedTools } from '../worker/mcp-provider'; import { statelessMcpServer } from '../workflow/mcp-client'; @@ -71,3 +72,39 @@ test('statelessMcpServer keeps cacheToolsList assignable and reports false with filtered.cacheToolsList = true; t.false(filtered.cacheToolsList); }); + +function serverWithResolver(onConnect: () => void): MCPServer { + return { + cacheToolsList: false, + name: 'tickets', + toolMetaResolver: () => ({ tenant: 'tenant-7' }), + async connect() { + onConnect(); + throw new Error('connection refused'); + }, + async close() {}, + async listTools() { + return []; + }, + async callTool() { + return []; + }, + async invalidateToolsCache() {}, + }; +} + +test('StatelessMCPServerProvider refuses a toolMetaResolver without connecting', async (t) => { + let connects = 0; + const activities = new StatelessMCPServerProvider('tickets', () => + serverWithResolver(() => connects++) + )._getActivities(); + + for (const name of ['tickets-stateless-list-tools', 'tickets-stateless-call-tool']) { + const err = await t.throwsAsync(activities[name]!({ toolName: 'echo', args: {} }), { + instanceOf: ApplicationFailure, + }); + t.is(err?.type, TOOL_META_RESOLVER_UNSUPPORTED_TYPE); + t.true(err?.nonRetryable); + } + t.is(connects, 0); +}); diff --git a/typescript/openai-agents/src/__tests__/test-openai-agents.ts b/typescript/openai-agents/src/__tests__/test-openai-agents.ts index 338fa424..3f4f034c 100644 --- a/typescript/openai-agents/src/__tests__/test-openai-agents.ts +++ b/typescript/openai-agents/src/__tests__/test-openai-agents.ts @@ -35,6 +35,8 @@ import { mcpFactoryArgWorkflow, mcpToolFilterResumeWorkflow, mcpToolFilterWorkflow, + mcpToolMetaWorkflow, + statefulMcpFactoryResolverWorkflow, summaryOverrideStringWorkflow, statefulMcpNotConnectedWorkflow, statefulMcpIsolationWorkflow, @@ -198,8 +200,9 @@ test('factoryArgument is passed through to MCP activities', async (t) => { }, ]; }, - async callTool(_toolName: string, _args: Record | null) { + async callTool(...callArgs: unknown[]) { t.deepEqual(factoryArgument, { tenantId: 'tenant-42' }, 'factoryArgument should be passed to callTool'); + t.is(callArgs.length, 2, 'callTool should get no meta argument when there is no toolMetaResolver'); return [{ type: 'text', text: '2026-01-01' }]; }, async invalidateToolsCache() {}, @@ -343,6 +346,104 @@ test('a tool filter that throws while resuming from a RunState fails the run', a }); }); +function metaTestServer(name: string, metas: unknown[]): MCPServer { + return { + cacheToolsList: false, + name, + async connect() {}, + async close() {}, + async listTools() { + return [ + { + name: 'echo_meta', + description: 'Records the request _meta', + inputSchema: { + type: 'object' as const, + properties: {}, + required: [] as string[], + additionalProperties: false, + }, + }, + ]; + }, + async callTool(_toolName: string, _args: Record | null, meta?: Record | null) { + metas.push(meta); + return [{ type: 'text', text: 'ok' }]; + }, + async invalidateToolsCache() {}, + }; +} + +test('MCP toolMetaResolver sends _meta on stateless and stateful servers', async (t) => { + const { createWorker, executeWorkflow } = helpers(t); + + const metas: unknown[] = []; + const worker = await createWorker({ + plugins: [ + new OpenAIAgentsPlugin({ + modelProvider: new FakeModelProvider([ + toolCallResponse('echo_meta', {}), + textResponse('stateless done'), + toolCallResponse('echo_meta', {}), + textResponse('stateful done'), + ]), + mcpServerProviders: [ + new StatelessMCPServerProvider('metaStateless', () => metaTestServer('metaStateless', metas)), + new StatefulMCPServerProvider( + 'metaStateful', + () => metaTestServer('metaStateful', metas), + t.context.env.nativeConnection + ), + ], + }), + ], + }); + + await worker.runUntil(async () => { + t.is( + await executeWorkflow(mcpToolMetaWorkflow, { args: [false], workflowExecutionTimeout: '30 seconds' }), + 'stateless done' + ); + t.is( + await executeWorkflow(mcpToolMetaWorkflow, { args: [true], workflowExecutionTimeout: '30 seconds' }), + 'stateful done' + ); + }); + + t.deepEqual(metas, [ + { tenant: 'tenant-7', tool: 'echo_meta', server: 'metaStateless' }, + { tenant: 'tenant-7', tool: 'echo_meta', server: 'metaStateful' }, + ]); +}); + +test('Stateful MCP: a factory toolMetaResolver fails the operations', async (t) => { + const { createWorker, startWorkflow } = helpers(t); + + const worker = await createWorker({ + plugins: [ + new OpenAIAgentsPlugin({ + modelProvider: new FakeModelProvider([]), + mcpServerProviders: [ + new StatefulMCPServerProvider( + 'factoryResolver', + () => ({ ...metaTestServer('factoryResolver', []), toolMetaResolver: () => ({ tenant: 'tenant-7' }) }), + t.context.env.nativeConnection + ), + ], + }), + ], + }); + + await worker.runUntil(async () => { + const handle = await startWorkflow(statefulMcpFactoryResolverWorkflow, { workflowExecutionTimeout: '20 seconds' }); + t.deepEqual(await handle.result(), [ + 'MCPToolMetaResolverUnsupported:true', + 'MCPToolMetaResolverUnsupported:true', + 'cleanup-success', + ]); + }); +}); + test('summary string is passed through to model activity', async (t) => { const { createWorker, taskQueue } = helpers(t); diff --git a/typescript/openai-agents/src/__tests__/workflows/openai-agents.ts b/typescript/openai-agents/src/__tests__/workflows/openai-agents.ts index cd0c8e2e..9d70bd09 100644 --- a/typescript/openai-agents/src/__tests__/workflows/openai-agents.ts +++ b/typescript/openai-agents/src/__tests__/workflows/openai-agents.ts @@ -2,10 +2,11 @@ import { Agent, getGlobalTraceProvider, type MCPToolFilterContext, + type MCPToolMetaContext, withTrace, withTraceContext, } from '@openai/agents-core'; -import { ApplicationFailure, proxyActivities } from '@temporalio/workflow'; +import { ActivityFailure, ApplicationFailure, proxyActivities } from '@temporalio/workflow'; import { WorkflowStream } from '@temporalio/workflow-streams/workflow'; import { TemporalOpenAIRunner, statelessMcpServer, statefulMcpServer } from '../../workflow'; @@ -158,6 +159,56 @@ export async function mcpToolFilterResumeWorkflow(): Promise { return 'resumed'; } +export async function mcpToolMetaWorkflow(stateful: boolean): Promise { + const toolMetaResolver = ({ runContext, toolName, serverName }: MCPToolMetaContext) => ({ + tenant: (runContext.context as { tenant: string }).tenant, + tool: toolName, + server: serverName, + }); + const server = stateful + ? statefulMcpServer('metaStateful', { toolMetaResolver }) + : statelessMcpServer('metaStateless', { toolMetaResolver }); + await server.connect(); + try { + const agent = new Agent({ + name: 'McpToolMetaAgent', + instructions: 'You have access to MCP tools.', + model: 'gpt-4o-mini', + mcpServers: [server], + }); + const result = await new TemporalOpenAIRunner().run(agent, 'Hello', { context: { tenant: 'tenant-7' } }); + return result.finalOutput ?? ''; + } finally { + await server.close(); + } +} + +function failureType(err: unknown): string { + const failure = err instanceof ActivityFailure ? err.cause : err; + return failure instanceof ApplicationFailure ? `${failure.type}:${failure.nonRetryable}` : String(err); +} + +export async function statefulMcpFactoryResolverWorkflow(): Promise { + const server = statefulMcpServer('factoryResolver'); + await server.connect(); + const results: string[] = []; + for (const operation of [() => server.listTools(), () => server.callTool('echo', {})]) { + try { + await operation(); + results.push('unexpected-success'); + } catch (err: unknown) { + results.push(failureType(err)); + } + } + try { + await server.cleanup(); + results.push('cleanup-success'); + } catch (err: unknown) { + results.push(failureType(err)); + } + return results; +} + export async function summaryOverrideStringWorkflow(prompt: string): Promise { const agent = new Agent({ name: 'SummaryAgent', diff --git a/typescript/openai-agents/src/common/mcp-types.ts b/typescript/openai-agents/src/common/mcp-types.ts index f8f03d2f..6729885a 100644 --- a/typescript/openai-agents/src/common/mcp-types.ts +++ b/typescript/openai-agents/src/common/mcp-types.ts @@ -1,4 +1,4 @@ -import type { MCPServer, MCPToolFilterCallable, MCPToolFilterStatic } from '@openai/agents-core'; +import type { MCPServer, MCPToolFilterCallable, MCPToolFilterStatic, MCPToolMetaResolver } from '@openai/agents-core'; export const MCP_LIST_TOOLS_SUFFIX = '-list-tools'; export const MCP_CALL_TOOL_SUFFIX = '-call-tool'; @@ -12,6 +12,9 @@ export const DEDICATED_WORKER_FAILURE_TYPE = 'DedicatedWorkerFailure'; /** Error type for a callable `toolFilter` on the server the factory returns, which the Worker refuses. */ export const TOOL_FILTER_UNSUPPORTED_TYPE = 'MCPToolFilterUnsupported'; +/** Error type for a `toolMetaResolver` on the server the factory returns, which the Worker refuses. */ +export const TOOL_META_RESOLVER_UNSUPPORTED_TYPE = 'MCPToolMetaResolverUnsupported'; + /** * Stateful variant of `MCPServer`. Adds `cleanup()` — the documented * lifecycle method users invoke from `finally` blocks after `connect()`. @@ -28,6 +31,14 @@ export interface McpServerHandleOptions { * registered with, and must be deterministic. */ toolFilter?: MCPToolFilterCallable | MCPToolFilterStatic; + /** + * Builds each tool call's MCP `_meta` in the Workflow, as for a native MCP server. It gets the + * run context and must be deterministic. Older plugin versions ignore this option and may not + * replay a History in which it ran, because running it can change the Workflow's commands or + * their order. Its result is recorded in Workflow History as part of the call-tool Activity + * input, so it must not hold secrets. + */ + toolMetaResolver?: MCPToolMetaResolver; } // Forwarded per-call (not stored on the provider) so different plugins sharing diff --git a/typescript/openai-agents/src/index.ts b/typescript/openai-agents/src/index.ts index 48fa99fa..03d2eefd 100644 --- a/typescript/openai-agents/src/index.ts +++ b/typescript/openai-agents/src/index.ts @@ -12,7 +12,7 @@ export type { ModelSummaryProvider, SerializableModelActivityOptions, } from './common/model-activity-options'; -export { DEDICATED_WORKER_FAILURE_TYPE, TOOL_FILTER_UNSUPPORTED_TYPE } from './common/mcp-types'; +export { DEDICATED_WORKER_FAILURE_TYPE, TOOL_FILTER_UNSUPPORTED_TYPE, TOOL_META_RESOLVER_UNSUPPORTED_TYPE } from './common/mcp-types'; export type { HostedToolCredentials, HostedToolCredentialsResolver, diff --git a/typescript/openai-agents/src/worker/mcp-provider.ts b/typescript/openai-agents/src/worker/mcp-provider.ts index 408d8837..bb608411 100644 --- a/typescript/openai-agents/src/worker/mcp-provider.ts +++ b/typescript/openai-agents/src/worker/mcp-provider.ts @@ -5,6 +5,7 @@ import { MCP_LIST_TOOLS_SUFFIX, MCP_STATELESS_SUFFIX, TOOL_FILTER_UNSUPPORTED_TYPE, + TOOL_META_RESOLVER_UNSUPPORTED_TYPE, } from '../common/mcp-types'; /** @@ -38,6 +39,33 @@ export function filterListedTools(server: MCPServer, return tools.filter((tool) => (allowed.length === 0 || allowed.includes(tool.name)) && !blocked.includes(tool.name)); } +/** + * Fails on a `toolMetaResolver` on the server the factory returns. It needs the run context, which + * exists only in the Workflow, so the Worker cannot run it. + */ +export function rejectToolMetaResolver(server: MCPServer): void { + if (server.toolMetaResolver) { + throw ApplicationFailure.create({ + message: + `MCP server '${server.name}' has a toolMetaResolver, which the Worker cannot run because the run ` + + 'context it gets exists only in the Workflow. Pass it as the toolMetaResolver option of ' + + 'statelessMcpServer() or statefulMcpServer() instead.', + type: TOOL_META_RESOLVER_UNSUPPORTED_TYPE, + nonRetryable: true, + }); + } +} + +/** Calls a tool with the `_meta` the Workflow built, as the agent does for a native server. */ +export async function callToolWithMeta( + server: MCPServer, + toolName: string, + args: Record | null, + meta: Record | undefined +): ReturnType { + return meta === undefined ? server.callTool(toolName, args) : server.callTool(toolName, args, meta); +} + export class StatelessMCPServerProvider { private readonly _name: string; @@ -53,6 +81,7 @@ export class StatelessMCPServerProvider { [`${this._name}${MCP_LIST_TOOLS_SUFFIX}`]: async (input?: { factoryArgument?: unknown }) => { const s = this.serverFactory(input?.factoryArgument); rejectCallableToolFilter(s); + rejectToolMetaResolver(s); try { await s.connect(); return filterListedTools(s, await s.listTools()); @@ -63,13 +92,15 @@ export class StatelessMCPServerProvider { [`${this._name}${MCP_CALL_TOOL_SUFFIX}`]: async (input: { toolName: string; args: Record | null; + meta?: Record; factoryArgument?: unknown; }) => { const s = this.serverFactory(input.factoryArgument); rejectCallableToolFilter(s); + rejectToolMetaResolver(s); try { await s.connect(); - return await s.callTool(input.toolName, input.args); + return await callToolWithMeta(s, input.toolName, input.args, input.meta); } finally { await s.close(); } diff --git a/typescript/openai-agents/src/worker/stateful-mcp-provider.ts b/typescript/openai-agents/src/worker/stateful-mcp-provider.ts index a54d0f37..89796f74 100644 --- a/typescript/openai-agents/src/worker/stateful-mcp-provider.ts +++ b/typescript/openai-agents/src/worker/stateful-mcp-provider.ts @@ -10,7 +10,7 @@ import { type StatefulMcpServerSessionArgs, } from '../common/mcp-types'; import { startAdaptiveHeartbeat } from './heartbeat'; -import { filterListedTools, rejectCallableToolFilter } from './mcp-provider'; +import { callToolWithMeta, filterListedTools, rejectCallableToolFilter, rejectToolMetaResolver } from './mcp-provider'; export function registerCancelShutdown(ctx: Context, dedicatedWorker: { shutdown: () => void }): void { ctx.cancelled.catch(() => { @@ -21,6 +21,7 @@ export function registerCancelShutdown(ctx: Context, dedicatedWorker: { shutdown interface CallToolArgs { toolName: string; args: Record | null; + meta?: Record; } /** @@ -74,6 +75,7 @@ export class StatefulMCPServerProvider { }); } rejectCallableToolFilter(server); + rejectToolMetaResolver(server); return server; }; @@ -83,7 +85,8 @@ export class StatefulMCPServerProvider { return filterListedTools(server, await server.listTools()); }; - const callTool = async (input: CallToolArgs) => getServerOrThrow().callTool(input.toolName, input.args); + const callTool = async (input: CallToolArgs) => + callToolWithMeta(getServerOrThrow(), input.toolName, input.args, input.meta); // Long-running session Activity on the MAIN Worker's Task Queue. Creates // a server, connects it, spins up a dedicated per-run Worker, then awaits diff --git a/typescript/openai-agents/src/workflow.ts b/typescript/openai-agents/src/workflow.ts index 263385d2..47aacedd 100644 --- a/typescript/openai-agents/src/workflow.ts +++ b/typescript/openai-agents/src/workflow.ts @@ -12,7 +12,7 @@ export type { AgentAsToolOptions } from './workflow/agent-tools'; export { statelessMcpServer } from './workflow/mcp-client'; export type { StatelessMcpServerOptions } from './workflow/mcp-client'; export type { StatefulTemporalMCPServer } from './common/mcp-types'; -export { DEDICATED_WORKER_FAILURE_TYPE, TOOL_FILTER_UNSUPPORTED_TYPE } from './common/mcp-types'; +export { DEDICATED_WORKER_FAILURE_TYPE, TOOL_FILTER_UNSUPPORTED_TYPE, TOOL_META_RESOLVER_UNSUPPORTED_TYPE } from './common/mcp-types'; export { statefulMcpServer } from './workflow/stateful-mcp-client'; export type { StatefulMcpServerOptions } from './workflow/stateful-mcp-client'; export { temporalSandboxClient } from './workflow/sandbox-client'; diff --git a/typescript/openai-agents/src/workflow/mcp-client.ts b/typescript/openai-agents/src/workflow/mcp-client.ts index 79b68146..6b23753e 100644 --- a/typescript/openai-agents/src/workflow/mcp-client.ts +++ b/typescript/openai-agents/src/workflow/mcp-client.ts @@ -45,6 +45,7 @@ export function statelessMcpServer(name: string, options?: StatelessMcpServerOpt sdkCache = value; }, toolFilter: options?.toolFilter, + toolMetaResolver: options?.toolMetaResolver, get name() { return name; }, @@ -64,10 +65,11 @@ export function statelessMcpServer(name: string, options?: StatelessMcpServerOpt } return tools; }, - async callTool(toolName: string, args: Record | null) { + async callTool(toolName: string, args: Record | null, meta?: Record | null) { return activities[callToolActivityName]!({ toolName, args, + ...(meta != null ? { meta } : {}), ...(factoryArg !== undefined ? { factoryArgument: factoryArg } : {}), }); }, diff --git a/typescript/openai-agents/src/workflow/stateful-mcp-client.ts b/typescript/openai-agents/src/workflow/stateful-mcp-client.ts index c951a0df..d54ebf40 100644 --- a/typescript/openai-agents/src/workflow/stateful-mcp-client.ts +++ b/typescript/openai-agents/src/workflow/stateful-mcp-client.ts @@ -1,4 +1,4 @@ -import type { MCPToolFilterCallable, MCPToolFilterStatic } from '@openai/agents-core'; +import type { MCPToolFilterCallable, MCPToolFilterStatic, MCPToolMetaResolver } from '@openai/agents-core'; import { ActivityFailure, ApplicationFailure, @@ -95,6 +95,7 @@ export class StatefulMCPServerReference implements StatefulTemporalMCPServer { private _connected = false; private _cachedTools: any[] | undefined; toolFilter?: MCPToolFilterCallable | MCPToolFilterStatic; + toolMetaResolver?: MCPToolMetaResolver; constructor(name: string, options?: StatefulMcpServerOptions) { this._name = `${name}${MCP_STATEFUL_SUFFIX}`; @@ -117,6 +118,9 @@ export class StatefulMCPServerReference implements StatefulTemporalMCPServer { typeof toolFilter === 'function' ? (context, tool) => toolFilter({ ...context, serverName: name }, tool) : toolFilter; + // Give the resolver the name the caller passed, as statelessMcpServer() does, not the suffixed one. + const toolMetaResolver = options?.toolMetaResolver; + this.toolMetaResolver = toolMetaResolver && ((context) => toolMetaResolver({ ...context, serverName: name })); } get name(): string { @@ -200,10 +204,18 @@ export class StatefulMCPServerReference implements StatefulTemporalMCPServer { return tools; } - async callTool(toolName: string, args: Record | null): Promise { + async callTool( + toolName: string, + args: Record | null, + meta?: Record | null + ): Promise { this.assertConnected(); return handleWorkerFailure(() => - scheduleActivity(this._name + MCP_CALL_TOOL_SUFFIX, [{ toolName, args }], this._operationConfig) + scheduleActivity( + this._name + MCP_CALL_TOOL_SUFFIX, + [{ toolName, args, ...(meta != null ? { meta } : {}) }], + this._operationConfig + ) ); }