diff --git a/typescript/openai-agents/README.md b/typescript/openai-agents/README.md index 3e727d1f..9fc4275c 100644 --- a/typescript/openai-agents/README.md +++ b/typescript/openai-agents/README.md @@ -511,6 +511,24 @@ export async function statefulMcpWorkflow(prompt: string): Promise { Dedicated Worker startup and heartbeat failures surface as `ApplicationFailure` type (exported as `DEDICATED_WORKER_FAILURE_TYPE`). +### MCP tool filters + +A static `toolFilter` on the server your factory returns, such as `createMCPToolStaticFilter({ blocked: ['delete_file'] })`, applies in the list-tools Activity, so the model is not offered a blocked tool. The filter applies only to the tool list: a run that listed tools before the Worker was upgraded keeps that list while it is cached. + +A callable filter gets the agent and the run context, which exist only in the Workflow, so the Worker refuses one on the factory's server: every list-tools and call-tool Activity for that server fails with a non-retryable `ApplicationFailure` of type `MCPToolFilterUnsupported` (exported as `TOOL_FILTER_UNSUPPORTED_TYPE`). A new run 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 callable filter from the factory's server before you upgrade Workers. Pass it as the `toolFilter` option of `statelessMcpServer()` or `statefulMcpServer()` instead: + +```ts +const server = statelessMcpServer('files', { + toolFilter: async ({ agent }, tool) => tool.name !== 'delete_file' || agent.name === 'FileAdmin', +}); +``` + +The agent applies this option in the Workflow each time it lists tools; the handle caches the server's full tool list, not the filtered one. A callable filter runs for each tool on every listing, gets the name the server was registered with as `serverName`, and must be deterministic. An Activity or timer it awaits adds commands that older plugin versions never schedule, so let Workflows that use the option finish before you roll the plugin back. A filter assigned to a stateful handle later (`server.toolFilter = ...`) gets the handle's internal name, `-stateful`. A callable filter also runs when 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 run fails with an `AgentsWorkflowError`. + +Because the handle caches the full list, the agent SDK's `invalidateServerToolsCache(name)` does not refresh a handle that has a filter. Call the handle's `invalidateToolsCache()` as well, which also works on older plugin versions. On older versions, `invalidateServerToolsCache(name)` alone refreshed a stateful handle with a `toolFilter` property assigned. Such a run schedules one list-tools Activity fewer on this version and fails replay, so let it finish before you upgrade. + +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. + ## Sandbox agents > **Experimental** — sandbox support is subject to change prior to General Availability. diff --git a/typescript/openai-agents/src/__tests__/mcp-provider.test.ts b/typescript/openai-agents/src/__tests__/mcp-provider.test.ts new file mode 100644 index 00000000..57234285 --- /dev/null +++ b/typescript/openai-agents/src/__tests__/mcp-provider.test.ts @@ -0,0 +1,73 @@ +import test from 'ava'; +import type { MCPServer } from '@openai/agents-core'; +import { ApplicationFailure } from '@temporalio/common'; +import { TOOL_FILTER_UNSUPPORTED_TYPE } from '../common/mcp-types'; +import { StatelessMCPServerProvider, filterListedTools } from '../worker/mcp-provider'; +import { statelessMcpServer } from '../workflow/mcp-client'; + +function fakeServer(toolFilter: MCPServer['toolFilter'], onConnect = () => {}): MCPServer { + return { + cacheToolsList: false, + name: 'files', + toolFilter, + async connect() { + onConnect(); + }, + async close() {}, + async listTools() { + return []; + }, + async callTool() { + return []; + }, + async invalidateToolsCache() {}, + }; +} + +test('filterListedTools applies a static filter with the agent SDK rules', (t) => { + const tools = ['read', 'read_all', 'delete'].map((name) => ({ name })); + const names = (toolFilter: MCPServer['toolFilter']) => + filterListedTools(fakeServer(toolFilter), tools).map((tool) => tool.name); + + t.deepEqual(names(undefined), ['read', 'read_all', 'delete']); + t.deepEqual(names({ allowedToolNames: [], blockedToolNames: [] }), ['read', 'read_all', 'delete']); + // Names match whole, not as prefixes. + t.deepEqual(names({ allowedToolNames: ['read'] }), ['read']); + t.deepEqual(names({ blockedToolNames: ['read'] }), ['read_all', 'delete']); + // A blocked name stays blocked when it is also allowed. + t.deepEqual(names({ allowedToolNames: ['read', 'delete'], blockedToolNames: ['delete'] }), ['read']); +}); + +test('StatelessMCPServerProvider refuses a callable toolFilter without connecting', async (t) => { + let connected = false; + const provider = new StatelessMCPServerProvider('files', () => + fakeServer( + async () => true, + () => { + connected = true; + } + ) + ); + const activities = provider._getActivities(); + + for (const name of ['files-stateless-list-tools', 'files-stateless-call-tool']) { + const err = await t.throwsAsync(activities[name]!({ toolName: 'read', args: {} }), { + instanceOf: ApplicationFailure, + }); + t.is(err?.type, TOOL_FILTER_UNSUPPORTED_TYPE); + t.true(err?.nonRetryable); + } + t.false(connected); +}); + +test('statelessMcpServer keeps cacheToolsList assignable and reports false with a filter', (t) => { + const server = statelessMcpServer('files'); + t.true(server.cacheToolsList); + server.cacheToolsList = false; + t.false(server.cacheToolsList); + + const filtered = statelessMcpServer('files', { toolFilter: { blockedToolNames: ['delete'] } }); + t.false(filtered.cacheToolsList); + filtered.cacheToolsList = true; + t.false(filtered.cacheToolsList); +}); diff --git a/typescript/openai-agents/src/__tests__/stubs/openai-agents-fakes.ts b/typescript/openai-agents/src/__tests__/stubs/openai-agents-fakes.ts index c72420d8..9db13832 100644 --- a/typescript/openai-agents/src/__tests__/stubs/openai-agents-fakes.ts +++ b/typescript/openai-agents/src/__tests__/stubs/openai-agents-fakes.ts @@ -10,6 +10,8 @@ import { export class FakeModel implements Model { private getNext: () => ModelResponse; + /** Every request the model got, in order. */ + readonly requests: ModelRequest[] = []; constructor(source: ModelResponse[] | Generator) { if (Array.isArray(source)) { @@ -37,7 +39,8 @@ export class FakeModel implements Model { } } - async getResponse(_request: ModelRequest): Promise { + async getResponse(request: ModelRequest): Promise { + this.requests.push(request); return this.getNext(); } diff --git a/typescript/openai-agents/src/__tests__/test-openai-agents.ts b/typescript/openai-agents/src/__tests__/test-openai-agents.ts index e57a9a7b..833df568 100644 --- a/typescript/openai-agents/src/__tests__/test-openai-agents.ts +++ b/typescript/openai-agents/src/__tests__/test-openai-agents.ts @@ -2,6 +2,7 @@ // the sibling `*.test.ts` files. import { addTraceProcessor, + createMCPToolStaticFilter, getCurrentTrace, getGlobalTraceProvider, setTracingDisabled, @@ -9,21 +10,26 @@ import { withTrace, } from '@openai/agents-core'; import type { MCPServer, Model, ModelProvider, Span, SpanData, TracingProcessor } from '@openai/agents-core'; -import { WorkflowClient } from '@temporalio/client'; +import { WorkflowClient, WorkflowFailedError } from '@temporalio/client'; +import type { ApplicationFailure } from '@temporalio/common'; import { temporal } from '@temporalio/proto'; import { helpers } from '@temporalio/test-helpers'; import { bundleWorkflowCode } from '@temporalio/worker'; +import type { NativeConnection } from '@temporalio/worker'; import { WorkflowStreamClient } from '@temporalio/workflow-streams/client'; import { OpenAIAgentsPlugin, OpenAIAgentsTraceClientInterceptor, StatefulMCPServerProvider, StatelessMCPServerProvider, + TOOL_FILTER_UNSUPPORTED_TYPE, type TracingApiKeyContext, } from '..'; import { localActivityAgentWorkflow, mcpFactoryArgWorkflow, + mcpToolFilterResumeWorkflow, + mcpToolFilterWorkflow, summaryOverrideStringWorkflow, statefulMcpNotConnectedWorkflow, statefulMcpIsolationWorkflow, @@ -35,7 +41,7 @@ import { tracingApiKeyWorkflow, } from './workflows/openai-agents'; import { makeTestFunction } from './helpers/test-fn'; -import { FakeModelProvider, textResponse, toolCallResponse } from './stubs/openai-agents'; +import { type FakeModel, FakeModelProvider, textResponse, toolCallResponse } from './stubs/openai-agents'; import { StreamingFakeModelProvider, streamingTextEvents } from './stubs/openai-agents-fakes'; import EventType = temporal.api.enums.v1.EventType; @@ -215,6 +221,122 @@ test('factoryArgument is passed through to MCP activities', async (t) => { }); }); +function filterTestServer(name: string, toolFilter: MCPServer['toolFilter']): MCPServer { + return { + cacheToolsList: false, + name, + toolFilter, + async connect() {}, + async close() {}, + async listTools() { + return ['safe_tool', 'danger_tool', 'extra_tool'].map((toolName) => ({ + name: toolName, + description: toolName, + inputSchema: { + type: 'object' as const, + properties: {}, + required: [] as string[], + additionalProperties: false, + }, + })); + }, + async callTool() { + return [{ type: 'text', text: 'unused' }]; + }, + async invalidateToolsCache() {}, + }; +} + +function filterTestProviders(toolFilter: MCPServer['toolFilter'], nativeConnection: NativeConnection) { + return [ + new StatelessMCPServerProvider('filterStateless', () => filterTestServer('filterStateless', toolFilter)), + new StatefulMCPServerProvider('filterStateful', () => filterTestServer('filterStateful', toolFilter), nativeConnection), + ]; +} + +test('MCP tool filters apply on stateless and stateful servers', async (t) => { + const { createWorker, startWorkflow } = helpers(t); + + const modelProvider = new FakeModelProvider(Array.from({ length: 6 }, () => textResponse('done'))); + const worker = await createWorker({ + plugins: [ + new OpenAIAgentsPlugin({ + modelProvider, + mcpServerProviders: filterTestProviders( + createMCPToolStaticFilter({ blocked: ['danger_tool'] }), + t.context.env.nativeConnection + ), + }), + ], + }); + + await worker.runUntil(async () => { + for (const [stateful, listToolsActivity] of [ + [false, 'filterStateless-stateless-list-tools'], + [true, 'filterStateful-stateful-list-tools'], + ] as const) { + const handle = await startWorkflow(mcpToolFilterWorkflow, { args: [stateful], workflowExecutionTimeout: '30 seconds' }); + t.deepEqual(await handle.result(), ['done', 'done', 'done']); + // The handle lists the server's tools once; the agent filters that list in every run. + const { events } = await handle.fetchHistory(); + const listings = events?.filter( + (e) => e.activityTaskScheduledEventAttributes?.activityType?.name === listToolsActivity + ); + t.is(listings?.length, 1); + } + }); + + // `danger_tool` is blocked by the factory server's static filter. The Workflow's filter allows + // `extra_tool` only in the first run, whose context is admin. + const offered = (modelProvider.getModel() as FakeModel).requests.map((r) => r.tools.map((tool) => tool.name)); + const perWorkflow = [['safe_tool', 'extra_tool'], ['safe_tool'], ['safe_tool']]; + t.deepEqual(offered, [...perWorkflow, ...perWorkflow]); +}); + +test('a callable toolFilter on the factory server fails the run', async (t) => { + const { createWorker, executeWorkflow } = helpers(t); + + const worker = await createWorker({ + plugins: [ + new OpenAIAgentsPlugin({ + modelProvider: new FakeModelProvider([]), + mcpServerProviders: filterTestProviders(async () => true, t.context.env.nativeConnection), + }), + ], + }); + + await worker.runUntil(async () => { + for (const stateful of [false, true]) { + const err = await t.throwsAsync( + executeWorkflow(mcpToolFilterWorkflow, { args: [stateful], workflowExecutionTimeout: '30 seconds' }), + { instanceOf: WorkflowFailedError } + ); + t.is((err?.cause?.cause as ApplicationFailure | undefined)?.type, TOOL_FILTER_UNSUPPORTED_TYPE); + } + }); +}); + +test('a tool filter that throws while resuming from a RunState fails the run', async (t) => { + const { createWorker, executeWorkflow } = helpers(t); + + const worker = await createWorker({ + plugins: [ + new OpenAIAgentsPlugin({ + modelProvider: new FakeModelProvider([textResponse('done')]), + mcpServerProviders: filterTestProviders(undefined, t.context.env.nativeConnection), + }), + ], + }); + + await worker.runUntil(async () => { + const err = await t.throwsAsync( + executeWorkflow(mcpToolFilterResumeWorkflow, { workflowExecutionTimeout: '15 seconds' }), + { instanceOf: WorkflowFailedError } + ); + t.is((err?.cause as ApplicationFailure | undefined)?.type, 'AgentsWorkflowError'); + }); +}); + 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 c32a76f1..1e162f27 100644 --- a/typescript/openai-agents/src/__tests__/workflows/openai-agents.ts +++ b/typescript/openai-agents/src/__tests__/workflows/openai-agents.ts @@ -1,4 +1,4 @@ -import { Agent } from '@openai/agents-core'; +import { Agent, type MCPToolFilterContext } from '@openai/agents-core'; import { ApplicationFailure } from '@temporalio/workflow'; import { WorkflowStream } from '@temporalio/workflow-streams/workflow'; import { TemporalOpenAIRunner, statelessMcpServer, statefulMcpServer } from '../../workflow'; @@ -99,6 +99,59 @@ export async function mcpFactoryArgWorkflow(prompt: string): Promise { return result.finalOutput ?? ''; } +export async function mcpToolFilterWorkflow(stateful: boolean): Promise { + const name = stateful ? 'filterStateful' : 'filterStateless'; + // The factory's server blocks `danger_tool` with a static filter; this filter blocks `extra_tool` + // unless the run's context allows it. + const toolFilter = async ({ runContext, serverName }: MCPToolFilterContext, tool: { name: string }) => + serverName === name && (tool.name !== 'extra_tool' || (runContext.context as { admin: boolean }).admin); + const server = stateful ? statefulMcpServer(name, { toolFilter }) : statelessMcpServer(name, { toolFilter }); + await server.connect(); + try { + const outputs: string[] = []; + // Two runs of one agent with different contexts, then a second agent on the same server. + for (const [agentName, admin] of [ + ['FirstAgent', true], + ['FirstAgent', false], + ['SecondAgent', false], + ] as const) { + const agent = new Agent({ + name: agentName, + instructions: 'You have access to MCP tools.', + model: 'gpt-4o-mini', + mcpServers: [server], + }); + const result = await new TemporalOpenAIRunner().run(agent, 'Hello', { context: { admin } }); + outputs.push(result.finalOutput ?? ''); + } + return outputs; + } finally { + await server.close(); + } +} + +class FilterSession { + isAdmin(): boolean { + return true; + } +} + +export async function mcpToolFilterResumeWorkflow(): Promise { + // Resuming from a RunState rebuilds the context from JSON, so `session` loses `isAdmin()`. + const toolFilter = async ({ runContext }: MCPToolFilterContext) => + (runContext.context as { session: FilterSession }).session.isAdmin(); + const agent = new Agent({ + name: 'FirstAgent', + instructions: 'You have access to MCP tools.', + model: 'gpt-4o-mini', + mcpServers: [statelessMcpServer('filterStateless', { toolFilter })], + }); + const runner = new TemporalOpenAIRunner(); + const result = await runner.run(agent, 'Hello', { context: { session: new FilterSession() } }); + await runner.run(agent, result.state); + return 'resumed'; +} + 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 c872916e..f8f03d2f 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 } from '@openai/agents-core'; +import type { MCPServer, MCPToolFilterCallable, MCPToolFilterStatic } from '@openai/agents-core'; export const MCP_LIST_TOOLS_SUFFIX = '-list-tools'; export const MCP_CALL_TOOL_SUFFIX = '-call-tool'; @@ -9,6 +9,9 @@ export const MCP_STATELESS_SUFFIX = '-stateless'; /** Error type for dedicated-Worker scheduling and heartbeat failures. */ 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'; + /** * Stateful variant of `MCPServer`. Adds `cleanup()` — the documented * lifecycle method users invoke from `finally` blocks after `connect()`. @@ -17,6 +20,16 @@ export interface StatefulTemporalMCPServer extends MCPServer { cleanup(): Promise; } +/** Options both MCP handles take. The agent reads them from the handle in the Workflow, as from a native MCP server. */ +export interface McpServerHandleOptions { + /** + * Tool filter the agent applies in the Workflow each time it lists tools, as for a native MCP + * server. A callable filter gets the agent, the run context and the name the server was + * registered with, and must be deterministic. + */ + toolFilter?: MCPToolFilterCallable | MCPToolFilterStatic; +} + // Forwarded per-call (not stored on the provider) so different plugins sharing // a provider each get their own per-run config. export interface StatefulMcpActivityInterceptorOptions { diff --git a/typescript/openai-agents/src/index.ts b/typescript/openai-agents/src/index.ts index 184507f1..48fa99fa 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 } from './common/mcp-types'; +export { DEDICATED_WORKER_FAILURE_TYPE, TOOL_FILTER_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 859b1544..408d8837 100644 --- a/typescript/openai-agents/src/worker/mcp-provider.ts +++ b/typescript/openai-agents/src/worker/mcp-provider.ts @@ -1,5 +1,42 @@ import type { MCPServer } from '@openai/agents-core'; -import { MCP_CALL_TOOL_SUFFIX, MCP_LIST_TOOLS_SUFFIX, MCP_STATELESS_SUFFIX } from '../common/mcp-types'; +import { ApplicationFailure } from '@temporalio/common'; +import { + MCP_CALL_TOOL_SUFFIX, + MCP_LIST_TOOLS_SUFFIX, + MCP_STATELESS_SUFFIX, + TOOL_FILTER_UNSUPPORTED_TYPE, +} from '../common/mcp-types'; + +/** + * Fails on a callable `toolFilter` on the server the factory returns. It needs the agent and the + * run context, which exist only in the Workflow, so the Worker cannot apply it. + */ +export function rejectCallableToolFilter(server: MCPServer): void { + if (typeof server.toolFilter === 'function') { + throw ApplicationFailure.create({ + message: + `MCP server '${server.name}' has a callable toolFilter, which the Worker cannot run because the agent ` + + 'and run context it gets exist only in the Workflow. Pass it as the toolFilter option of ' + + 'statelessMcpServer() or statefulMcpServer() instead.', + type: TOOL_FILTER_UNSUPPORTED_TYPE, + nonRetryable: true, + }); + } +} + +/** + * Applies the static `toolFilter` of the server the factory returns, as the agent does for a native + * server. A callable filter is refused by `rejectCallableToolFilter()` before this runs. + */ +export function filterListedTools(server: MCPServer, tools: T[]): T[] { + const filter = server.toolFilter; + if (!filter || typeof filter === 'function') { + return tools; + } + const allowed = filter.allowedToolNames ?? []; + const blocked = filter.blockedToolNames ?? []; + return tools.filter((tool) => (allowed.length === 0 || allowed.includes(tool.name)) && !blocked.includes(tool.name)); +} export class StatelessMCPServerProvider { private readonly _name: string; @@ -15,9 +52,10 @@ export class StatelessMCPServerProvider { return { [`${this._name}${MCP_LIST_TOOLS_SUFFIX}`]: async (input?: { factoryArgument?: unknown }) => { const s = this.serverFactory(input?.factoryArgument); + rejectCallableToolFilter(s); try { await s.connect(); - return await s.listTools(); + return filterListedTools(s, await s.listTools()); } finally { await s.close(); } @@ -28,6 +66,7 @@ export class StatelessMCPServerProvider { factoryArgument?: unknown; }) => { const s = this.serverFactory(input.factoryArgument); + rejectCallableToolFilter(s); try { await s.connect(); return await s.callTool(input.toolName, input.args); diff --git a/typescript/openai-agents/src/worker/stateful-mcp-provider.ts b/typescript/openai-agents/src/worker/stateful-mcp-provider.ts index 55331bb7..a54d0f37 100644 --- a/typescript/openai-agents/src/worker/stateful-mcp-provider.ts +++ b/typescript/openai-agents/src/worker/stateful-mcp-provider.ts @@ -10,6 +10,7 @@ import { type StatefulMcpServerSessionArgs, } from '../common/mcp-types'; import { startAdaptiveHeartbeat } from './heartbeat'; +import { filterListedTools, rejectCallableToolFilter } from './mcp-provider'; export function registerCancelShutdown(ctx: Context, dedicatedWorker: { shutdown: () => void }): void { ctx.cancelled.catch(() => { @@ -72,11 +73,15 @@ export class StatefulMCPServerProvider { nonRetryable: true, }); } + rejectCallableToolFilter(server); return server; }; // Operation Activities — registered on the per-run Task Queue. - const listTools = async () => getServerOrThrow().listTools(); + const listTools = async () => { + const server = getServerOrThrow(); + return filterListedTools(server, await server.listTools()); + }; const callTool = async (input: CallToolArgs) => getServerOrThrow().callTool(input.toolName, input.args); diff --git a/typescript/openai-agents/src/workflow.ts b/typescript/openai-agents/src/workflow.ts index 6a2dd998..263385d2 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 } from './common/mcp-types'; +export { DEDICATED_WORKER_FAILURE_TYPE, TOOL_FILTER_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 f944b46d..79b68146 100644 --- a/typescript/openai-agents/src/workflow/mcp-client.ts +++ b/typescript/openai-agents/src/workflow/mcp-client.ts @@ -1,9 +1,14 @@ import type { MCPServer } from '@openai/agents-core'; import type { Duration, RetryPolicy } from '@temporalio/common'; import { proxyActivities } from '@temporalio/workflow'; -import { MCP_CALL_TOOL_SUFFIX, MCP_LIST_TOOLS_SUFFIX, MCP_STATELESS_SUFFIX } from '../common/mcp-types'; +import { + MCP_CALL_TOOL_SUFFIX, + MCP_LIST_TOOLS_SUFFIX, + MCP_STATELESS_SUFFIX, + type McpServerHandleOptions, +} from '../common/mcp-types'; -export interface StatelessMcpServerOptions { +export interface StatelessMcpServerOptions extends McpServerHandleOptions { cacheToolsList?: boolean; startToCloseTimeout?: Duration; heartbeatTimeout?: Duration; @@ -26,10 +31,20 @@ export function statelessMcpServer(name: string, options?: StatelessMcpServerOpt let cachedTools: any[] | undefined; const shouldCache = options?.cacheToolsList ?? true; + // What the agent SDK sees. Assigning `cacheToolsList` changes only this, as it did before. + let sdkCache = shouldCache; const factoryArg = options?.factoryArgument; return { - cacheToolsList: shouldCache, + // The agent SDK caches the filtered list, so with a filter only this handle caches, and the + // agent filters every listing. + get cacheToolsList() { + return sdkCache && !this.toolFilter; + }, + set cacheToolsList(value: boolean) { + sdkCache = value; + }, + toolFilter: options?.toolFilter, get name() { return name; }, diff --git a/typescript/openai-agents/src/workflow/runner.ts b/typescript/openai-agents/src/workflow/runner.ts index 732cfedc..e74c4442 100644 --- a/typescript/openai-agents/src/workflow/runner.ts +++ b/typescript/openai-agents/src/workflow/runner.ts @@ -282,7 +282,10 @@ export class TemporalOpenAIRunner { // agent (with ActivityBackedModel); setCurrentAgent only swaps the top-level ref, not nested ones. // It also drops upstream's live-session identity, which would otherwise resolve manifest // environment references on the Workflow thread rather than in the resume Activity. - const restored = (await RunState.fromString(converted, input.toString())) as RunState; + // fromString lists tools, which runs MCP tool filters; their errors fail the run, as in the run below. + const restored = (await RunState.fromString(converted, input.toString()).catch((error: unknown) => { + throw normalizeAgentsRunError(error); + })) as RunState; // Suppress upstream Runner.run's withTrace(state._trace, ...) branch; we want the current Workflow's trace. restored.clearTrace(); preparedInput = restored; diff --git a/typescript/openai-agents/src/workflow/stateful-mcp-client.ts b/typescript/openai-agents/src/workflow/stateful-mcp-client.ts index e23e6592..c951a0df 100644 --- a/typescript/openai-agents/src/workflow/stateful-mcp-client.ts +++ b/typescript/openai-agents/src/workflow/stateful-mcp-client.ts @@ -1,3 +1,4 @@ +import type { MCPToolFilterCallable, MCPToolFilterStatic } from '@openai/agents-core'; import { ActivityFailure, ApplicationFailure, @@ -19,12 +20,13 @@ import { MCP_LIST_TOOLS_SUFFIX, MCP_SERVER_SESSION_SUFFIX, MCP_STATEFUL_SUFFIX, + type McpServerHandleOptions, type StatefulMcpServerSessionArgs, type StatefulTemporalMCPServer, } from '../common/mcp-types'; import { getCurrentPluginConfig } from './plugin-config-store'; -export interface StatefulMcpServerOptions { +export interface StatefulMcpServerOptions extends McpServerHandleOptions { /** Activity options for tool operation Activities on the dedicated Worker. */ config?: { startToCloseTimeout?: Duration; @@ -40,9 +42,9 @@ export interface StatefulMcpServerOptions { }; factoryArgument?: unknown; /** - * When `true` (the default), the agent SDK caches `listTools` for the - * agent's lifetime. Set to `false` if the server's tool set may change - * mid-run. + * When `true` (the default), the tool list is fetched once and reused for + * the rest of the Workflow run. Set to `false` if the server's tool set may + * change mid-run. */ cacheToolsList?: boolean; } @@ -91,6 +93,8 @@ export class StatefulMCPServerReference implements StatefulTemporalMCPServer { private _sessionScope: CancellationScope | undefined; private _sessionPromise: Promise | undefined; private _connected = false; + private _cachedTools: any[] | undefined; + toolFilter?: MCPToolFilterCallable | MCPToolFilterStatic; constructor(name: string, options?: StatefulMcpServerOptions) { this._name = `${name}${MCP_STATEFUL_SUFFIX}`; @@ -107,14 +111,22 @@ export class StatefulMCPServerReference implements StatefulTemporalMCPServer { }; this._factoryArgument = options?.factoryArgument; this._cacheToolsList = options?.cacheToolsList ?? true; + // A callable filter gets the registered name, as with statelessMcpServer(), not `this.name`. + const toolFilter = options?.toolFilter; + this.toolFilter = + typeof toolFilter === 'function' + ? (context, tool) => toolFilter({ ...context, serverName: name }, tool) + : toolFilter; } get name(): string { return this._name; } + // The agent SDK caches the filtered list, so with a filter only this handle caches, and the + // agent filters every listing. get cacheToolsList(): boolean { - return this._cacheToolsList; + return this._cacheToolsList && !this.toolFilter; } /** Starts the per-run server-session Activity. */ @@ -175,7 +187,17 @@ export class StatefulMCPServerReference implements StatefulTemporalMCPServer { async listTools(): Promise { this.assertConnected(); - return handleWorkerFailure(() => scheduleActivity(this._name + MCP_LIST_TOOLS_SUFFIX, [], this._operationConfig)); + const shouldCache = this._cacheToolsList && !!this.toolFilter; + if (shouldCache && this._cachedTools) { + return this._cachedTools; + } + const tools = await handleWorkerFailure(() => + scheduleActivity(this._name + MCP_LIST_TOOLS_SUFFIX, [], this._operationConfig) + ); + if (shouldCache) { + this._cachedTools = tools; + } + return tools; } async callTool(toolName: string, args: Record | null): Promise { @@ -186,8 +208,9 @@ export class StatefulMCPServerReference implements StatefulTemporalMCPServer { } async invalidateToolsCache(): Promise { - // Caching is owned by the upstream agent SDK; nothing to invalidate on the - // Workflow-side handle. + // Without a filter, caching is owned by the upstream agent SDK. With one, this cache holds the + // full list, and the agent SDK's invalidateServerToolsCache(name) cannot clear it. + this._cachedTools = undefined; } }