diff --git a/api/server/controllers/mcp.js b/api/server/controllers/mcp.js index 61d6a6bdda3..1eb1b6eb3d0 100644 --- a/api/server/controllers/mcp.js +++ b/api/server/controllers/mcp.js @@ -219,13 +219,17 @@ const getMCPTools = async (req, res) => { let serverTools; let publicationGeneration; + let publicationRevision; try { - ({ tools: serverTools, publicationGeneration } = - await mcpManager.getServerToolFunctionsSnapshot( - userId, - serverName, - mcpConfig[serverName], - )); + ({ + tools: serverTools, + publicationGeneration, + publicationRevision, + } = await mcpManager.getServerToolFunctionsSnapshot( + userId, + serverName, + mcpConfig[serverName], + )); } catch (error) { logger.error(`[getMCPTools] Error fetching tools for server ${serverName}:`, error); continue; @@ -243,6 +247,7 @@ const getMCPTools = async (req, res) => { serverTools, serverConfig: mcpConfig[serverName], publicationGeneration, + publicationRevision, }).catch((err) => logger.error(`[getMCPTools] Failed to cache tools for ${serverName}:`, err), ); diff --git a/api/server/services/Tools/mcp.js b/api/server/services/Tools/mcp.js index 15606944190..78325d77208 100644 --- a/api/server/services/Tools/mcp.js +++ b/api/server/services/Tools/mcp.js @@ -66,6 +66,7 @@ async function reinitMCPServer({ let oauthExpiresAt; let ephemeralServer = false; let publicationGeneration; + let publicationRevision; try { const registry = getMCPServersRegistry(); @@ -279,6 +280,20 @@ async function reinitMCPServer({ } if (snapshot.complete) { tools = snapshot.tools; + /** Reserved before this snapshot's tools/list; an app-level catalog cannot publish + * without it, and allocating a later one here would outrank fresher tools. */ + publicationRevision = snapshot.publicationRevision; + if (snapshot.orderingUnavailable && typeof connection.refreshToolList === 'function') { + /** These tools still serve this request; the connection republishes the shared + * catalog under backoff rather than leaving it cold until the next reinitialize. */ + connection + .refreshToolList() + .catch((err) => + logger.debug( + `[MCP Reinitialize] Could not schedule a catalog republish for ${serverName}: ${err?.message ?? String(err)}`, + ), + ); + } } else { logger.warn( `[MCP Reinitialize] Preserving cached tools for ${serverName} because tools/list returned an incomplete snapshot`, @@ -306,6 +321,7 @@ async function reinitMCPServer({ tools, serverConfig, ...(publicationGeneration && { publicationGeneration }), + ...(publicationRevision && { publicationRevision }), }); if (availableTools == null) { tools = null; diff --git a/api/server/services/Tools/mcp.spec.js b/api/server/services/Tools/mcp.spec.js index 65029bd68b3..f4e6b2629fb 100644 --- a/api/server/services/Tools/mcp.spec.js +++ b/api/server/services/Tools/mcp.spec.js @@ -125,6 +125,50 @@ describe('reinitMCPServer — customUserVars gating (issue #10969)', () => { }); }); + /** An app-level catalog write is dropped unless it carries the ordering reserved before its + * own tools/list. When this path forwarded no revision, every publication was discarded and + * agents were told the server had no tools at all (#14857). */ + it('publishes under the ordering its snapshot was fetched with', async () => { + mockGetConnection.mockResolvedValue({ + fetchOrderedToolsSnapshot: jest.fn().mockResolvedValue({ + tools: [{ name: 'search', inputSchema: { type: 'object' } }], + complete: true, + publicationRevision: '7', + }), + }); + + await reinitMCPServer({ + user, + serverName, + serverConfig: { type: 'streamable-http', url: 'https://thingy.example.com/mcp' }, + }); + + expect(mockUpdateMCPServerTools).toHaveBeenCalledWith( + expect.objectContaining({ serverName, publicationRevision: '7' }), + ); + }); + + it('asks the connection to republish a catalog it could not order', async () => { + const refreshToolList = jest.fn().mockResolvedValue(undefined); + mockGetConnection.mockResolvedValue({ + refreshToolList, + fetchOrderedToolsSnapshot: jest.fn().mockResolvedValue({ + tools: [{ name: 'search', inputSchema: { type: 'object' } }], + complete: true, + orderingUnavailable: true, + }), + }); + + const result = await reinitMCPServer({ + user, + serverName, + serverConfig: { type: 'streamable-http', url: 'https://thingy.example.com/mcp' }, + }); + + expect(refreshToolList).toHaveBeenCalledTimes(1); + expect(result.tools).toHaveLength(1); + }); + it('preserves cached tools when live recovery returns an incomplete snapshot', async () => { const fetchOrderedToolsSnapshot = jest.fn().mockResolvedValue({ tools: [{ name: 'partial', inputSchema: { type: 'object' } }], diff --git a/packages/api/src/mcp/ConnectionsRepository.ts b/packages/api/src/mcp/ConnectionsRepository.ts index 63e9396192d..84d67c9f60b 100644 --- a/packages/api/src/mcp/ConnectionsRepository.ts +++ b/packages/api/src/mcp/ConnectionsRepository.ts @@ -162,18 +162,30 @@ export class ConnectionsRepository { this.connections.set(serverName, connection); if (this.ownerId === undefined && options.refreshTools !== false) { + /** The snapshot carries ordering reserved before its own `tools/list`, so this + * first-connect publication is ordered against concurrent replicas exactly as a + * list_changed refresh is. An app-level write that cannot be ordered is dropped, which + * left agents with a permanently empty catalog when this path populated it (#14857). */ if (connection.client.getServerCapabilities()?.tools == null) { + const ordering = await connection.reserveToolsPublicationRevision(); + /** The refresh path reserves again under backoff. Publishing unordered instead would be + * dropped in silence, leaving whatever this server last advertised in place. */ + if (ordering.orderingUnavailable) { + await connection.refreshToolList(); + return connection; + } await notifyMCPToolsChanged({ tools: [], serverName, serverConfig, publicationGeneration, + publicationRevision: ordering.publicationRevision, }); return connection; } const initialGeneration = toolsChangedGeneration; const snapshot = await connection.fetchToolsSnapshot(); - if (snapshot.complete) { + if (snapshot.complete && !snapshot.orderingUnavailable) { if (toolsChangedGeneration !== initialGeneration) { await latestToolsChangedPublication; } else { @@ -182,6 +194,7 @@ export class ConnectionsRepository { serverName, serverConfig, publicationGeneration, + publicationRevision: snapshot.publicationRevision, }); } } else { diff --git a/packages/api/src/mcp/MCPManager.ts b/packages/api/src/mcp/MCPManager.ts index 6741ded3aa1..df6a1d02acb 100644 --- a/packages/api/src/mcp/MCPManager.ts +++ b/packages/api/src/mcp/MCPManager.ts @@ -422,6 +422,7 @@ export class MCPManager extends UserConnectionManager { ): Promise<{ tools: t.LCAvailableTools | null; publicationGeneration?: string; + publicationRevision?: string; }> { try { const registry = MCPServersRegistry.getInstance(); @@ -434,9 +435,7 @@ export class MCPManager extends UserConnectionManager { ? await this.appConnections?.get(serverName) : null; if (existingAppConnection != null) { - return { - tools: await MCPServerInspector.getToolFunctions(serverName, existingAppConnection), - }; + return MCPServerInspector.getToolCatalog(serverName, existingAppConnection); } let awaitedRecovery: Promise | undefined; @@ -482,7 +481,7 @@ export class MCPManager extends UserConnectionManager { } try { - const tools = await MCPServerInspector.getToolFunctions(serverName, connection); + const { tools } = await MCPServerInspector.getToolCatalog(serverName, connection); const generationAfterFetch = await getMCPToolsChangedGeneration({ userId, serverName }); if ( publicationGeneration != null && diff --git a/packages/api/src/mcp/__tests__/ConnectionsRepository.test.ts b/packages/api/src/mcp/__tests__/ConnectionsRepository.test.ts index 5105df396ec..98ebd7f41ba 100644 --- a/packages/api/src/mcp/__tests__/ConnectionsRepository.test.ts +++ b/packages/api/src/mcp/__tests__/ConnectionsRepository.test.ts @@ -1,6 +1,10 @@ import { logger } from '@librechat/data-schemas'; import type * as t from '~/mcp/types'; -import { getMCPAppToolsPublicationGeneration, setMCPToolsChangedHandler } from '~/mcp/toolsChanged'; +import { + setMCPToolsChangedHandler, + setMCPToolsChangedRevisionHandler, + getMCPAppToolsPublicationGeneration, +} from '~/mcp/toolsChanged'; import { ConnectionsRepository } from '~/mcp/ConnectionsRepository'; import { MCPConnectionFactory } from '~/mcp/MCPConnectionFactory'; import { MCPConnection } from '~/mcp/connection'; @@ -83,6 +87,7 @@ describe('ConnectionsRepository', () => { disconnect: jest.fn().mockResolvedValue(undefined), dispose: jest.fn().mockResolvedValue(undefined), fetchToolsSnapshot: jest.fn().mockResolvedValue({ tools: [], complete: true }), + reserveToolsPublicationRevision: jest.fn().mockResolvedValue({}), refreshToolList: jest.fn().mockResolvedValue(undefined), createdAt: Date.now(), isStale: jest.fn().mockReturnValue(false), @@ -101,6 +106,7 @@ describe('ConnectionsRepository', () => { afterEach(() => { setMCPToolsChangedHandler(null); + setMCPToolsChangedRevisionHandler(null); jest.clearAllMocks(); }); @@ -185,6 +191,71 @@ describe('ConnectionsRepository', () => { await expect(load).resolves.toBe(mockConnection); }); + /** An app-level write that carries no revision cannot be ordered and is dropped, so the + * ordering reserved for this snapshot has to reach the publication (#14857). */ + it('publishes the initial app snapshot under the revision it was fetched with', async () => { + mockConnection.fetchToolsSnapshot.mockResolvedValue({ + tools: [], + complete: true, + publicationRevision: '4', + }); + const handler = jest.fn(); + setMCPToolsChangedHandler(handler); + + await repository.get('server1'); + + expect(handler).toHaveBeenCalledWith( + expect.objectContaining({ serverName: 'server1', publicationRevision: '4' }), + ); + }); + + it('reserves ordering for a server that advertises no tools capability', async () => { + (mockConnection.client.getServerCapabilities as jest.Mock).mockReturnValue({}); + mockConnection.reserveToolsPublicationRevision = jest + .fn() + .mockResolvedValue({ publicationRevision: '9' }); + const handler = jest.fn(); + setMCPToolsChangedHandler(handler); + + await repository.get('server1'); + + expect(mockConnection.fetchToolsSnapshot).not.toHaveBeenCalled(); + expect(handler).toHaveBeenCalledWith( + expect.objectContaining({ serverName: 'server1', tools: [], publicationRevision: '9' }), + ); + }); + + /** An unordered publication is dropped in silence, so a server left with no way to order + * its empty catalog has to retry rather than leave the previous one advertised. */ + it('retries an empty catalog it could not reserve ordering for', async () => { + (mockConnection.client.getServerCapabilities as jest.Mock).mockReturnValue({}); + mockConnection.reserveToolsPublicationRevision = jest + .fn() + .mockResolvedValue({ orderingUnavailable: true }); + const handler = jest.fn(); + setMCPToolsChangedHandler(handler); + + await repository.get('server1'); + + expect(mockConnection.refreshToolList).toHaveBeenCalledTimes(1); + expect(handler).not.toHaveBeenCalled(); + }); + + it('retries a fetched catalog it could not reserve ordering for', async () => { + mockConnection.fetchToolsSnapshot.mockResolvedValue({ + tools: [], + complete: true, + orderingUnavailable: true, + }); + const handler = jest.fn(); + setMCPToolsChangedHandler(handler); + + await repository.get('server1'); + + expect(mockConnection.refreshToolList).toHaveBeenCalledTimes(1); + expect(handler).not.toHaveBeenCalled(); + }); + it('can defer the initial app tool refresh for startup synchronization', async () => { await repository.get('server1', { refreshTools: false }); diff --git a/packages/api/src/mcp/__tests__/MCPManager.test.ts b/packages/api/src/mcp/__tests__/MCPManager.test.ts index 9a07b7e6f47..f9c02e7bb63 100644 --- a/packages/api/src/mcp/__tests__/MCPManager.test.ts +++ b/packages/api/src/mcp/__tests__/MCPManager.test.ts @@ -532,7 +532,7 @@ describe('MCPManager', () => { describe('getServerToolFunctions', () => { it('should catch and handle errors gracefully', async () => { - (MCPServerInspector.getToolFunctions as jest.Mock) = jest.fn(() => { + (MCPServerInspector.getToolCatalog as jest.Mock) = jest.fn(() => { throw new Error('Connection failed'); }); @@ -552,7 +552,7 @@ describe('MCPManager', () => { }); it('should catch synchronous errors from getUserConnections', async () => { - (MCPServerInspector.getToolFunctions as jest.Mock) = jest.fn().mockResolvedValue({}); + (MCPServerInspector.getToolCatalog as jest.Mock) = jest.fn().mockResolvedValue({ tools: {} }); mockAppConnections({ get: jest.fn().mockResolvedValue(null), @@ -586,9 +586,9 @@ describe('MCPManager', () => { }, }; - (MCPServerInspector.getToolFunctions as jest.Mock) = jest + (MCPServerInspector.getToolCatalog as jest.Mock) = jest .fn() - .mockResolvedValue(expectedTools); + .mockResolvedValue({ tools: expectedTools }); mockAppConnections({ has: jest.fn().mockResolvedValue(true), @@ -608,7 +608,9 @@ describe('MCPManager', () => { finishInspection = resolve; }); const connection = {} as MCPConnection; - (MCPServerInspector.getToolFunctions as jest.Mock) = jest.fn().mockReturnValue(inspection); + (MCPServerInspector.getToolCatalog as jest.Mock) = jest + .fn() + .mockReturnValue(inspection.then((tools) => ({ tools }))); mockAppConnections({ get: jest.fn().mockResolvedValue(null), }); @@ -643,7 +645,7 @@ describe('MCPManager', () => { const recovery = new Promise((resolve) => { resolveRecovery = resolve; }); - (MCPServerInspector.getToolFunctions as jest.Mock) = jest.fn().mockResolvedValue({}); + (MCPServerInspector.getToolCatalog as jest.Mock) = jest.fn().mockResolvedValue({ tools: {} }); mockAppConnections({ get: jest.fn().mockResolvedValue(null), }); @@ -665,14 +667,14 @@ describe('MCPManager', () => { const toolsPromise = manager.getServerToolFunctions(userId, serverName); await new Promise((resolve) => setImmediate(resolve)); - expect(MCPServerInspector.getToolFunctions).not.toHaveBeenCalled(); + expect(MCPServerInspector.getToolCatalog).not.toHaveBeenCalled(); internals.userConnections.set(userId, new Map([[serverName, recoveredConnection]])); internals.oauthRecoveries.delete(staleConnection); resolveRecovery?.(); await expect(toolsPromise).resolves.toEqual({}); - expect(MCPServerInspector.getToolFunctions).toHaveBeenCalledWith( + expect(MCPServerInspector.getToolCatalog).toHaveBeenCalledWith( serverName, recoveredConnection, ); @@ -681,7 +683,7 @@ describe('MCPManager', () => { it('should include specific server name in error messages', async () => { const specificServerName = 'github_mcp_server'; - (MCPServerInspector.getToolFunctions as jest.Mock) = jest.fn(() => { + (MCPServerInspector.getToolCatalog as jest.Mock) = jest.fn(() => { throw new Error('Server specific error'); }); @@ -721,7 +723,7 @@ describe('MCPManager', () => { const appGet = jest.fn().mockResolvedValue({} as MCPConnection); mockAppConnections({ get: appGet }); (mockRegistryInstance.isAppServerConfig as jest.Mock).mockResolvedValue(false); - (MCPServerInspector.getToolFunctions as jest.Mock).mockResolvedValue(expectedTools); + (MCPServerInspector.getToolCatalog as jest.Mock).mockResolvedValue({ tools: expectedTools }); const manager = await MCPManager.createInstance(newMCPServersConfig()); const internals = manager as unknown as { @@ -733,10 +735,7 @@ describe('MCPManager', () => { manager.getServerToolFunctionsSnapshot(userId, serverName, overlayConfig), ).resolves.toEqual({ tools: expectedTools, publicationGeneration: undefined }); expect(appGet).not.toHaveBeenCalled(); - expect(MCPServerInspector.getToolFunctions).toHaveBeenCalledWith( - serverName, - overlayConnection, - ); + expect(MCPServerInspector.getToolCatalog).toHaveBeenCalledWith(serverName, overlayConnection); }); }); @@ -3468,7 +3467,9 @@ describe('MCPManager', () => { }, }, }; - (MCPServerInspector.getToolFunctions as jest.Mock).mockResolvedValue(expectedToolFunctions); + (MCPServerInspector.getToolCatalog as jest.Mock).mockResolvedValue({ + tools: expectedToolFunctions, + }); try { const manager = await MCPManager.createInstance(newMCPServersConfig()); @@ -3528,7 +3529,7 @@ describe('MCPManager', () => { await expect( manager.getServerToolFunctionsSnapshot(userId, serverName, serverConfig), ).resolves.toEqual({ tools: null }); - expect(MCPServerInspector.getToolFunctions).not.toHaveBeenCalled(); + expect(MCPServerInspector.getToolCatalog).not.toHaveBeenCalled(); expect(connection.dispose).toHaveBeenCalledTimes(1); } finally { generationSpy.mockRestore(); @@ -3568,7 +3569,7 @@ describe('MCPManager', () => { await expect( manager.getServerToolFunctionsSnapshot(userId, serverName, committedConfig), ).resolves.toEqual({ tools: null }); - expect(MCPServerInspector.getToolFunctions).not.toHaveBeenCalled(); + expect(MCPServerInspector.getToolCatalog).not.toHaveBeenCalled(); expect(connection.dispose).toHaveBeenCalledTimes(1); } finally { generationSpy.mockRestore(); @@ -3605,7 +3606,7 @@ describe('MCPManager', () => { await expect(manager.getServerToolFunctionsSnapshot(userId, serverName)).resolves.toEqual({ tools: null, }); - expect(MCPServerInspector.getToolFunctions).not.toHaveBeenCalled(); + expect(MCPServerInspector.getToolCatalog).not.toHaveBeenCalled(); expect(connection.dispose).toHaveBeenCalledTimes(1); } finally { generationSpy.mockRestore(); diff --git a/packages/api/src/mcp/__tests__/appToolPublication.integration.test.ts b/packages/api/src/mcp/__tests__/appToolPublication.integration.test.ts new file mode 100644 index 00000000000..016d745a2a0 --- /dev/null +++ b/packages/api/src/mcp/__tests__/appToolPublication.integration.test.ts @@ -0,0 +1,197 @@ +/** Real-SDK coverage for app-level catalogs that never reach agents (#14857). */ +import { Server } from '@modelcontextprotocol/sdk/server/index.js'; +import { InMemoryTransport } from '@modelcontextprotocol/sdk/inMemory.js'; +import { CacheKeys, Constants, normalizeServerName } from 'librechat-data-provider'; +import { ListToolsRequestSchema, type Tool } from '@modelcontextprotocol/sdk/types.js'; +import type { MCPToolsChangedEvent } from '../toolsChanged'; +import type { ParsedServerConfig } from '../types'; +import { + notifyMCPToolsChanged, + setMCPToolsChangedHandler, + getMCPAppToolsPublicationGeneration, + setMCPToolsChangedRevisionHandler, +} from '../toolsChanged'; +import { createMCPCatalogStore } from '../catalog/store'; +import { createMCPToolCacheService } from '../tools'; +import { MCPConnection } from '../connection'; + +jest.setTimeout(10_000); + +const SERVER_NAME = 'shared'; + +const appConfig: ParsedServerConfig = { + type: 'streamable-http', + url: 'https://mcp.example.com/mcp', +}; + +const tool = (name: string): Tool => ({ + name, + description: `${name} the corpus`, + inputSchema: { type: 'object', properties: {} }, +}); + +const toolKey = (name: string) => + `${name}${Constants.mcp_delimiter}${normalizeServerName(SERVER_NAME)}`; + +/** Wires the store, the cache service, and the publication handlers exactly as startup does. */ +function createHarness() { + const cache = new Map(); + const store = createMCPCatalogStore({ + cacheConfig: { FORCED_IN_MEMORY_CACHE_NAMESPACES: [CacheKeys.TOOL_CACHE] }, + getCache: () => ({ + get: async (key) => cache.get(key), + set: async (key, value) => { + cache.set(key, value); + return true; + }, + delete: async (key) => cache.delete(key), + }), + }); + + const service = createMCPToolCacheService({ + getCachedTools: store.getCachedTools, + updateCachedGlobalTools: store.updateCachedGlobalTools, + setCachedTools: store.setCachedTools, + setCachedToolsIfCurrent: store.setCachedToolsIfCurrent, + getCachedAppServerTools: store.getCachedAppServerTools, + setCachedAppServerTools: store.setCachedAppServerTools, + getServerConfig: async () => appConfig, + getAllServerConfigs: async () => ({ [SERVER_NAME]: appConfig }), + isAppServerConfig: async () => true, + }); + + setMCPToolsChangedRevisionHandler(({ serverName, configGeneration }) => + store.getNextAppToolsPublicationRevision(serverName, configGeneration), + ); + setMCPToolsChangedHandler(async (event: MCPToolsChangedEvent) => { + await service.updateMCPServerTools({ + userId: event.userId, + serverName: event.serverName, + tools: event.tools, + serverConfig: event.serverConfig as ParsedServerConfig, + publicationGeneration: event.publicationGeneration, + publicationRevision: event.publicationRevision, + }); + }); + + return { store, service }; +} + +async function createConnection(tools: Tool[]) { + const server = new Server( + { name: 'shared-tool-server', version: '1.0.0' }, + { capabilities: { tools: { listChanged: true } } }, + ); + server.setRequestHandler(ListToolsRequestSchema, async () => ({ + tools: tools.map((entry) => ({ ...entry })), + })); + + const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair(); + await server.connect(serverTransport); + + /** No userId: this is an app-shared connection, the scope agents read from. */ + const connection = new MCPConnection({ serverName: SERVER_NAME, serverConfig: appConfig }); + await connection.client.connect(clientTransport); + connection.emit('connectionChange', 'connected'); + + return { + connection, + notifyChanged: () => server.sendToolListChanged(), + close: async () => { + connection.removeAllListeners(); + await connection.client.close().catch(() => undefined); + await server.close().catch(() => undefined); + }, + }; +} + +describe('app-level tool publication', () => { + const configGeneration = getMCPAppToolsPublicationGeneration(appConfig); + let harness: Awaited> | undefined; + + afterEach(async () => { + setMCPToolsChangedHandler(null); + setMCPToolsChangedRevisionHandler(null); + await harness?.close(); + harness = undefined; + }); + + /** The reporter's case: a shared server connecting outside the startup refresh. Its snapshot + * has to carry ordering, or the catalog write is dropped and agents see no tools at all. */ + it('makes a first-connect snapshot readable by agents', async () => { + const { service } = createHarness(); + harness = await createConnection([tool('search')]); + + const snapshot = await harness.connection.fetchToolsSnapshot(); + expect(snapshot.publicationRevision).toBeDefined(); + + await notifyMCPToolsChanged({ + tools: snapshot.tools, + serverName: SERVER_NAME, + serverConfig: appConfig, + publicationGeneration: configGeneration, + publicationRevision: snapshot.publicationRevision, + }); + + const published = await service.getMCPServerTools('user-1', SERVER_NAME, appConfig); + expect(Object.keys(published ?? {})).toEqual([toolKey('search')]); + }); + + it('reserves ordering before the tools/list it describes', async () => { + const { store } = createHarness(); + harness = await createConnection([tool('search')]); + + const before = await store.getNextAppToolsPublicationRevision(SERVER_NAME, configGeneration); + const snapshot = await harness.connection.fetchToolsSnapshot(); + const after = await store.getNextAppToolsPublicationRevision(SERVER_NAME, configGeneration); + + expect(Number(snapshot.publicationRevision)).toBeGreaterThan(Number(before)); + expect(Number(snapshot.publicationRevision)).toBeLessThan(Number(after)); + }); + + /** A snapshot fetched earlier must not overwrite a catalog published from a later fetch, + * which is exactly what a revision allocated at publish time would allow. */ + it('cannot overwrite a catalog fetched after it', async () => { + const { service } = createHarness(); + harness = await createConnection([tool('stale')]); + const stale = await harness.connection.fetchToolsSnapshot(); + + const current = await harness.connection.reserveToolsPublicationRevision(); + await notifyMCPToolsChanged({ + tools: [tool('current')], + serverName: SERVER_NAME, + serverConfig: appConfig, + publicationGeneration: configGeneration, + publicationRevision: current.publicationRevision, + }); + await notifyMCPToolsChanged({ + tools: stale.tools, + serverName: SERVER_NAME, + serverConfig: appConfig, + publicationGeneration: configGeneration, + publicationRevision: stale.publicationRevision, + }); + + const published = await service.getMCPServerTools('user-1', SERVER_NAME, appConfig); + expect(Object.keys(published ?? {})).toEqual([toolKey('current')]); + }); + + /** When a request-time fetch is superseded by a list_changed refresh, it returns the refresh's + * catalog — and must return the refresh's revision with it, not its own superseded ticket. */ + it('carries the refresh revision when a request fetch is superseded', async () => { + createHarness(); + harness = await createConnection([tool('search')]); + const published: Array = []; + harness.connection.on('toolsChanged', (_tools: Tool[], revision?: string) => + published.push(revision), + ); + + await harness.notifyChanged(); + await harness.connection.refreshToolList(); + const ordered = await harness.connection.fetchOrderedToolsSnapshot(); + + expect(published.length).toBeGreaterThan(0); + expect(ordered.complete).toBe(true); + expect(ordered.publicationRevision).toBeDefined(); + }); +}); diff --git a/packages/api/src/mcp/assistants.ts b/packages/api/src/mcp/assistants.ts index 25c123d0fc1..6b0160ce4ac 100644 --- a/packages/api/src/mcp/assistants.ts +++ b/packages/api/src/mcp/assistants.ts @@ -29,6 +29,7 @@ export interface AssistantToolDefinitionsParams { export interface AssistantToolCatalogSnapshot { tools: LCAvailableTools | null; publicationGeneration?: string; + publicationRevision?: string; } export interface AssistantToolDefinitionsDeps { @@ -60,6 +61,7 @@ export interface AssistantToolDefinitionsDeps { serverTools: LCAvailableTools; serverConfig: ParsedServerConfig; publicationGeneration?: string; + publicationRevision?: string; }) => Promise; } @@ -134,6 +136,7 @@ async function loadServerCatalog( serverTools: snapshot.tools, serverConfig, publicationGeneration: snapshot.publicationGeneration, + publicationRevision: snapshot.publicationRevision, }) .catch((error) => logger.error( diff --git a/packages/api/src/mcp/connection.ts b/packages/api/src/mcp/connection.ts index 64eeed4c57f..9254a4db7aa 100644 --- a/packages/api/src/mcp/connection.ts +++ b/packages/api/src/mcp/connection.ts @@ -1002,6 +1002,11 @@ type MCPListToolsResult = Awaited>; export interface MCPToolsSnapshot { tools: MCPListToolsResult['tools']; complete: boolean; + /** Ordering ticket reserved before this snapshot's `tools/list`; app scope only. */ + publicationRevision?: string; + /** Set when reserving that ticket failed, which is retryable — unlike a scope that simply + * has no ordering to reserve, where the ticket is absent because none was ever needed. */ + orderingUnavailable?: boolean; } export class MCPConnection extends EventEmitter { @@ -1042,6 +1047,7 @@ export class MCPConnection extends EventEmitter { epoch: number; generation: number; tools: MCPListToolsResult['tools']; + publicationRevision?: string; } | null = null; private hasConnected = false; @@ -1853,25 +1859,9 @@ export class MCPConnection extends EventEmitter { const refreshEpoch = this.toolListRefreshEpoch; while (this.handledToolListChangeGeneration < this.toolListChangeGeneration) { const targetGeneration = this.toolListChangeGeneration; - let publicationRevision: string | undefined; - try { - publicationRevision = await reserveMCPToolsChangedRevision({ - serverName: this.serverName, - serverConfig: this.options, - userId: this.userId, - }); - } catch (error) { - this.toolListRefreshFailures++; - logger.error( - `${this.getLogPrefix()} Failed to reserve tool-list publication order:`, - error, - ); - this.scheduleToolListRefreshRetry(); - return; - } - const snapshot = + const snapshot: MCPToolsSnapshot = this.client.getServerCapabilities()?.tools == null - ? { tools: [], complete: true } + ? { tools: [], complete: true, ...(await this.reserveToolsPublicationRevision()) } : await this.fetchToolsSnapshot(); if ( this.toolListRefreshEpoch !== refreshEpoch || @@ -1880,7 +1870,8 @@ export class MCPConnection extends EventEmitter { ) { return; } - if (!snapshot.complete) { + /** Publishing unordered would drop this catalog silently; retry until it can be ordered. */ + if (!snapshot.complete || snapshot.orderingUnavailable) { this.toolListRefreshFailures++; this.scheduleToolListRefreshRetry(); return; @@ -1892,8 +1883,9 @@ export class MCPConnection extends EventEmitter { epoch: refreshEpoch, generation: targetGeneration, tools: snapshot.tools, + publicationRevision: snapshot.publicationRevision, }; - this.dispatchToolsChanged(snapshot.tools, publicationRevision); + this.dispatchToolsChanged(snapshot.tools, snapshot.publicationRevision); } } @@ -2390,11 +2382,20 @@ export class MCPConnection extends EventEmitter { const maxPages = mcpConfig.TOOLS_LIST_MAX_PAGES; const maxTools = mcpConfig.TOOLS_LIST_MAX_TOOLS; const maxBytes = mcpConfig.TOOLS_LIST_MAX_BYTES; + /** Reserved before the first page so the resulting catalog can never outrank one published + * from a `tools/list` that started later. Every app-level publisher reads its ordering off + * the snapshot it received, which is the only way to know when the data was actually read. */ + const ordering = await this.reserveToolsPublicationRevision(); const deadline = Date.now() + mcpConfig.TOOLS_LIST_TIMEOUT_MS; const allTools: MCPListToolsResult['tools'] = []; const seenCursors = new Set(); let cursor: string | undefined; let totalBytes = 0; + const snapshot = (complete: boolean): MCPToolsSnapshot => ({ + tools: allTools, + complete, + ...ordering, + }); for (let page = 1; page <= maxPages; page++) { const exhaustedBudget = getToolsListBudgetExceededReason( @@ -2405,31 +2406,31 @@ export class MCPConnection extends EventEmitter { ); if (exhaustedBudget != null) { this.warnToolsListBudgetExceeded(exhaustedBudget, allTools.length); - return { tools: allTools, complete: true }; + return snapshot(true); } const remainingMs = deadline - Date.now(); if (remainingMs <= 0) { this.warnToolsListBudgetExceeded('time', allTools.length); - return { tools: allTools, complete: false }; + return snapshot(false); } const result = await this.listToolsPage(cursor, remainingMs); if (result == null) { /** Request failed mid-pagination: return the pages already fetched instead of discarding them. */ - return { tools: allTools, complete: false }; + return snapshot(false); } for (const tool of result.tools) { if (allTools.length >= maxTools) { this.warnToolsListBudgetExceeded('tool count', allTools.length); - return { tools: allTools, complete: true }; + return snapshot(true); } const toolBytes = getApproximateToolBytes(tool); if (totalBytes + toolBytes > maxBytes) { this.warnToolsListBudgetExceeded('size', allTools.length); - return { tools: allTools, complete: true }; + return snapshot(true); } allTools.push(tool); @@ -2438,7 +2439,7 @@ export class MCPConnection extends EventEmitter { const { nextCursor } = result; if (nextCursor == null) { - return { tools: allTools, complete: true }; + return snapshot(true); } const nextPageBudget = getToolsListBudgetExceededReason( @@ -2449,14 +2450,14 @@ export class MCPConnection extends EventEmitter { ); if (nextPageBudget != null) { this.warnToolsListBudgetExceeded(nextPageBudget, allTools.length); - return { tools: allTools, complete: true }; + return snapshot(true); } if (seenCursors.has(nextCursor)) { logger.warn( `${this.getLogPrefix()} MCP server returned a repeated tools/list cursor; stopping pagination after ${page} page(s).`, ); - return { tools: allTools, complete: false }; + return snapshot(false); } seenCursors.add(nextCursor); @@ -2466,7 +2467,29 @@ export class MCPConnection extends EventEmitter { logger.warn( `${this.getLogPrefix()} Reached the tools/list pagination limit of ${maxPages} page(s); some tools may be omitted. Set MCP_TOOLS_LIST_MAX_PAGES higher if this server legitimately exposes more.`, ); - return { tools: allTools, complete: true }; + return snapshot(true); + } + + /** + * Allocates app-catalog ordering. A reservation failure must not fail the request that asked + * for the tools, so it is reported on the snapshot for publishers to retry on instead. + */ + public async reserveToolsPublicationRevision(): Promise<{ + publicationRevision?: string; + orderingUnavailable?: boolean; + }> { + try { + return { + publicationRevision: await reserveMCPToolsChangedRevision({ + serverName: this.serverName, + serverConfig: this.options, + userId: this.userId, + }), + }; + } catch (error) { + logger.warn(`${this.getLogPrefix()} Failed to reserve tool-list publication order:`, error); + return { orderingUnavailable: true }; + } } /** @@ -2507,7 +2530,13 @@ export class MCPConnection extends EventEmitter { published.generation === this.toolListChangeGeneration && this.handledToolListChangeGeneration === this.toolListChangeGeneration ) { - return { tools: published.tools, complete: true }; + /** Ordering travels with the data: this is the refresh's catalog, so it must publish under + * the refresh's revision rather than the one reserved for the superseded fetch above. */ + return { + tools: published.tools, + complete: true, + publicationRevision: published.publicationRevision, + }; } return { tools: [], complete: false }; diff --git a/packages/api/src/mcp/registry/MCPServerInspector.ts b/packages/api/src/mcp/registry/MCPServerInspector.ts index 3281cdf6bea..b0c1b436e0b 100644 --- a/packages/api/src/mcp/registry/MCPServerInspector.ts +++ b/packages/api/src/mcp/registry/MCPServerInspector.ts @@ -162,22 +162,22 @@ export class MCPServerInspector { } private async fetchToolFunctions(): Promise { - this.config.toolFunctions = await MCPServerInspector.getToolFunctions( - this.serverName, - this.connection!, - ); + this.config.toolFunctions = ( + await MCPServerInspector.getToolCatalog(this.serverName, this.connection!) + ).tools; } /** - * Converts server tools to LibreChat-compatible tool functions format. + * Converts server tools to LibreChat-compatible tool functions format, keeping the ordering + * reserved before the `tools/list` that produced them. App-level publishers need that + * revision — a catalog write that cannot be ordered against concurrent replicas is dropped. * @param serverName - The name of the server * @param connection - The MCP connection - * @returns Tool functions formatted for LibreChat */ - public static async getToolFunctions( + public static async getToolCatalog( serverName: string, connection: MCPConnection, - ): Promise { + ): Promise<{ tools: t.LCAvailableTools; publicationRevision?: string }> { const snapshot = await connection.fetchOrderedToolsSnapshot(); if (!snapshot.complete) { throw new Error(`Incomplete tools/list snapshot for MCP server ${serverName}`); @@ -206,6 +206,6 @@ export class MCPServerInspector { }; }); - return toolFunctions; + return { tools: toolFunctions, publicationRevision: snapshot.publicationRevision }; } } diff --git a/packages/api/src/mcp/registry/__tests__/MCPServerInspector.test.ts b/packages/api/src/mcp/registry/__tests__/MCPServerInspector.test.ts index b3f6aaf1f42..52990606991 100644 --- a/packages/api/src/mcp/registry/__tests__/MCPServerInspector.test.ts +++ b/packages/api/src/mcp/registry/__tests__/MCPServerInspector.test.ts @@ -501,7 +501,7 @@ describe('MCPServerInspector', () => { }); }); - describe('getToolFunctions()', () => { + describe('getToolCatalog()', () => { it('should convert MCP tools to LibreChat tool functions format', async () => { mockConnection.fetchOrderedToolsSnapshot = jest.fn().mockResolvedValue({ complete: true, @@ -528,7 +528,10 @@ describe('MCPServerInspector', () => { ], }); - const result = await MCPServerInspector.getToolFunctions('my_server', mockConnection); + const { tools: result } = await MCPServerInspector.getToolCatalog( + 'my_server', + mockConnection, + ); expect(result).toEqual({ file_read_mcp_my_server: { @@ -564,7 +567,10 @@ describe('MCPServerInspector', () => { .fn() .mockResolvedValue({ tools: [], complete: true }); - const result = await MCPServerInspector.getToolFunctions('my_server', mockConnection); + const { tools: result } = await MCPServerInspector.getToolCatalog( + 'my_server', + mockConnection, + ); expect(result).toEqual({}); }); @@ -581,7 +587,10 @@ describe('MCPServerInspector', () => { ], }); - const result = await MCPServerInspector.getToolFunctions('My Server', mockConnection); + const { tools: result } = await MCPServerInspector.getToolCatalog( + 'My Server', + mockConnection, + ); const key = 'file_read_mcp_My_Server'; expect(Object.keys(result)).toEqual([key]); @@ -594,9 +603,9 @@ describe('MCPServerInspector', () => { complete: false, }); - await expect( - MCPServerInspector.getToolFunctions('my_server', mockConnection), - ).rejects.toThrow('Incomplete tools/list snapshot for MCP server my_server'); + await expect(MCPServerInspector.getToolCatalog('my_server', mockConnection)).rejects.toThrow( + 'Incomplete tools/list snapshot for MCP server my_server', + ); }); }); }); diff --git a/packages/api/src/mcp/tools.spec.ts b/packages/api/src/mcp/tools.spec.ts index 9f314cff916..03726be831a 100644 --- a/packages/api/src/mcp/tools.spec.ts +++ b/packages/api/src/mcp/tools.spec.ts @@ -1,3 +1,4 @@ +import { logger } from '@librechat/data-schemas'; import { Constants, normalizeServerName } from 'librechat-data-provider'; import type { LCAvailableTools, ParsedServerConfig } from './types'; import type { MCPToolCacheDeps, MCPToolInput } from './tools'; @@ -157,6 +158,8 @@ describe('createMCPToolCacheService', () => { ).rejects.toThrow('Redis down'); }); + /** A publisher that lost its snapshot's revision fetched at an unknown time. Allocating a + * fresh one here would let a slow fetch of an old catalog outrank a newer one. */ it('does not publish a live app snapshot without pre-fetch ordering', async () => { const deps = createMockDeps(); @@ -171,6 +174,75 @@ describe('createMCPToolCacheService', () => { expect(deps.setCachedAppServerTools).not.toHaveBeenCalled(); }); + /** #14857 went a release without a diagnostic because dropping an app catalog only logged + * at debug. A drop means agents lose this server's tools, so it has to be visible by + * default; a superseded write is routine and must stay quiet. */ + describe('visibility of a discarded publication', () => { + const publish = (params: { publicationGeneration?: string; publicationRevision?: string }) => + createMCPToolCacheService( + createMockDeps({ setCachedAppServerTools: jest.fn().mockResolvedValue(false) }), + ).replaceAppServerTools({ serverName: 'dynamic', serverTools: {}, ...params }); + + let warn: jest.SpyInstance; + + beforeEach(() => { + warn = jest.spyOn(logger, 'warn').mockImplementation(() => logger); + }); + + afterEach(() => warn.mockRestore()); + + it('warns when a publication cannot be ordered', async () => { + await publish({ publicationGeneration: 'config-generation' }); + + expect(warn).toHaveBeenCalledWith(expect.stringContaining('Skipped unordered')); + }); + + it('warns when a publication cannot be addressed', async () => { + await publish({ publicationRevision: '1' }); + + expect(warn).toHaveBeenCalledWith(expect.stringContaining('Skipped unaddressed')); + }); + + it('stays quiet when a concurrent replica already published newer tools', async () => { + await publish({ publicationGeneration: 'config-generation', publicationRevision: '1' }); + + expect(warn).not.toHaveBeenCalled(); + }); + }); + + /** The catalog write needs ordering; the tools themselves were read from the server and are + * correct to serve. Discarding them is what surfaced as a server with no tools (#14857). */ + it('serves tools it could not publish instead of discarding them', async () => { + const deps = createSharedCacheDeps({ config: cacheableConfig }); + const search = toolName('search', 'dynamic'); + + await expect( + createMCPToolCacheService(deps).updateMCPServerTools({ + userId: 'user-1', + serverName: 'dynamic', + serverConfig: cacheableConfig, + tools: [{ name: 'search' }], + }), + ).resolves.toEqual({ [search]: expect.objectContaining({ type: 'function' }) }); + + expect(deps.setCachedAppServerTools).not.toHaveBeenCalled(); + }); + + it('discards a superseded catalog rather than serving it', async () => { + const deps = createSharedCacheDeps({ config: cacheableConfig }); + deps.setCachedAppServerTools = jest.fn().mockResolvedValue(false); + + await expect( + createMCPToolCacheService(deps).updateMCPServerTools({ + userId: 'user-1', + serverName: 'dynamic', + serverConfig: cacheableConfig, + tools: [{ name: 'search' }], + publicationRevision: '1', + }), + ).resolves.toBeNull(); + }); + it('rejects a tool boundary owned by another app server', async () => { const shadowed = toolName('search', 'foo_bar'); const deps = createMockDeps({ diff --git a/packages/api/src/mcp/tools.ts b/packages/api/src/mcp/tools.ts index 35dfd39fcd6..b1b801e1ca8 100644 --- a/packages/api/src/mcp/tools.ts +++ b/packages/api/src/mcp/tools.ts @@ -288,6 +288,17 @@ export function createMCPToolCacheService(deps: MCPToolCacheDeps): MCPToolCacheS userId == null ? (publicationGeneration ?? configGeneration) : (configGeneration ?? publicationGeneration); + /** Only the shared catalog write needs ordering. These tools were just read from the + * server, so the caller should still serve them; discarding a correct tool list because + * its write could not be ordered is what makes a cache failure look to the user like a + * server with no tools at all (#14857). A superseded write is different — another + * replica holds something newer — and still discards below. */ + if (!publicationRevision) { + logger.warn( + `[MCP Cache] Serving ${tools.length} unpublished tools for ${serverName}: this snapshot reserved no revision, so every request re-fetches them`, + ); + return serverTools; + } const replaced = await replaceAppServerTools({ serverName, serverTools, @@ -368,12 +379,23 @@ export function createMCPToolCacheService(deps: MCPToolCacheDeps): MCPToolCacheS const config = await resolveCacheConfig(undefined, serverName); configGeneration = config ? getMCPAppToolsPublicationGeneration(config) : undefined; } + /** Discarding a publication is warned, not debugged: #14857 was invisible for a release + * because the only trace of a dropped app catalog was a debug line no deployment runs. + * A drop here means this server's tools are missing for every agent that needs them. */ if (!configGeneration) { - logger.debug(`[MCP Cache] Skipped unaddressed app-level publication for ${serverName}`); + logger.warn( + `[MCP Cache] Skipped unaddressed app-level publication for ${serverName}; its tools stay unavailable to agents`, + ); return false; } + /** Ordering is reserved before the `tools/list` that produced these tools and travels with + * the snapshot, so a publisher that lost it fetched at an unknown time and cannot be + * ordered against concurrent replicas. Allocating one here instead would let a slow fetch + * of an old catalog outrank a newer one that reserved after it started. */ if (!publicationRevision) { - logger.debug(`[MCP Cache] Skipped unordered app-level publication for ${serverName}`); + logger.warn( + `[MCP Cache] Skipped unordered app-level publication for ${serverName}: its snapshot carried no reserved revision, so its tools stay unavailable to agents`, + ); return false; } const replaced = await setCachedAppServerTools( @@ -382,9 +404,10 @@ export function createMCPToolCacheService(deps: MCPToolCacheDeps): MCPToolCacheS serverTools, publicationRevision, ); + /** Expected whenever replicas publish concurrently: the winner already holds newer tools. */ if (replaced === false) { logger.debug( - `[MCP Cache] Ignored superseded app-level tools for ${serverName} at revision ${publicationRevision ?? '0'}`, + `[MCP Cache] Ignored superseded app-level tools for ${serverName} at revision ${publicationRevision}`, ); return false; }