mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-10-08 09:26:17 +00:00
🚐 fix: Reuse Request-Scoped MCP Connections per Run (#13673)
Some checks are pending
Docker Dev Branch Images Build / build (Dockerfile, lc-dev, node) (push) Waiting to run
Docker Dev Branch Images Build / build (Dockerfile.multi, lc-dev-api, api-build) (push) Waiting to run
GitNexus Index / index (push) Waiting to run
GitNexus Index / post-index (push) Blocked by required conditions
Docker Dev Images Build / build (Dockerfile, librechat-dev, node) (push) Waiting to run
Docker Dev Images Build / build (Dockerfile.multi, librechat-dev-api, api-build) (push) Waiting to run
Sync Locize Translations & Create Translation PR / Sync Translation Keys with Locize (push) Waiting to run
Sync Locize Translations & Create Translation PR / Create Translation PR on Version Published (push) Blocked by required conditions
Sync Helm Chart Tags / Ignore non-main push (push) Waiting to run
Sync Helm Chart Tags / Sync chart tags (push) Waiting to run
Some checks are pending
Docker Dev Branch Images Build / build (Dockerfile, lc-dev, node) (push) Waiting to run
Docker Dev Branch Images Build / build (Dockerfile.multi, lc-dev-api, api-build) (push) Waiting to run
GitNexus Index / index (push) Waiting to run
GitNexus Index / post-index (push) Blocked by required conditions
Docker Dev Images Build / build (Dockerfile, librechat-dev, node) (push) Waiting to run
Docker Dev Images Build / build (Dockerfile.multi, librechat-dev-api, api-build) (push) Waiting to run
Sync Locize Translations & Create Translation PR / Sync Translation Keys with Locize (push) Waiting to run
Sync Locize Translations & Create Translation PR / Create Translation PR on Version Published (push) Blocked by required conditions
Sync Helm Chart Tags / Ignore non-main push (push) Waiting to run
Sync Helm Chart Tags / Sync chart tags (push) Waiting to run
* fix(mcp): reuse request-scoped connections per run * test(mcp): update connection factory defaults
This commit is contained in:
parent
65bca95023
commit
139d61c437
20 changed files with 291 additions and 25 deletions
|
|
@ -348,6 +348,7 @@ const OpenAIChatCompletionController = async (req, res) => {
|
|||
* @type {Map<string, {
|
||||
* agent: object,
|
||||
* toolRegistry?: import('@librechat/agents').LCToolRegistry,
|
||||
* requestScopedConnections?: import('@librechat/api').RequestScopedMCPConnectionStore,
|
||||
* userMCPAuthMap?: Record<string, Record<string, string>>,
|
||||
* tool_resources?: object,
|
||||
* actionsEnabled?: boolean,
|
||||
|
|
@ -493,6 +494,7 @@ const OpenAIChatCompletionController = async (req, res) => {
|
|||
signal: abortController.signal,
|
||||
toolRegistry: ctx.toolRegistry,
|
||||
mcpAvailableTools: ctx.mcpAvailableTools,
|
||||
requestScopedConnections: ctx.requestScopedConnections,
|
||||
userMCPAuthMap: ctx.userMCPAuthMap,
|
||||
tool_resources: ctx.tool_resources,
|
||||
actionsEnabled: ctx.actionsEnabled,
|
||||
|
|
|
|||
|
|
@ -467,6 +467,7 @@ const createResponse = async (req, res) => {
|
|||
* @type {Map<string, {
|
||||
* agent: object,
|
||||
* toolRegistry?: import('@librechat/agents').LCToolRegistry,
|
||||
* requestScopedConnections?: import('@librechat/api').RequestScopedMCPConnectionStore,
|
||||
* userMCPAuthMap?: Record<string, Record<string, string>>,
|
||||
* tool_resources?: object,
|
||||
* actionsEnabled?: boolean,
|
||||
|
|
@ -685,6 +686,7 @@ const createResponse = async (req, res) => {
|
|||
signal: abortController.signal,
|
||||
toolRegistry: ctx.toolRegistry,
|
||||
mcpAvailableTools: ctx.mcpAvailableTools,
|
||||
requestScopedConnections: ctx.requestScopedConnections,
|
||||
userMCPAuthMap: ctx.userMCPAuthMap,
|
||||
tool_resources: ctx.tool_resources,
|
||||
actionsEnabled: ctx.actionsEnabled,
|
||||
|
|
@ -860,6 +862,7 @@ const createResponse = async (req, res) => {
|
|||
signal: abortController.signal,
|
||||
toolRegistry: ctx.toolRegistry,
|
||||
mcpAvailableTools: ctx.mcpAvailableTools,
|
||||
requestScopedConnections: ctx.requestScopedConnections,
|
||||
userMCPAuthMap: ctx.userMCPAuthMap,
|
||||
tool_resources: ctx.tool_resources,
|
||||
actionsEnabled: ctx.actionsEnabled,
|
||||
|
|
|
|||
|
|
@ -184,6 +184,7 @@ const initializeClient = async ({ req, res, signal, endpointOption }) => {
|
|||
* agent?: object,
|
||||
* tool_resources?: object,
|
||||
* toolRegistry?: import('@librechat/agents').LCToolRegistry,
|
||||
* requestScopedConnections?: import('@librechat/api').RequestScopedMCPConnectionStore,
|
||||
* openAIApiKey?: string
|
||||
* }>}
|
||||
*/
|
||||
|
|
@ -204,6 +205,7 @@ const initializeClient = async ({ req, res, signal, endpointOption }) => {
|
|||
agent: ctx.agent,
|
||||
toolRegistry: ctx.toolRegistry,
|
||||
mcpAvailableTools: ctx.mcpAvailableTools,
|
||||
requestScopedConnections: ctx.requestScopedConnections,
|
||||
userMCPAuthMap: ctx.userMCPAuthMap,
|
||||
tool_resources: ctx.tool_resources,
|
||||
actionsEnabled: ctx.actionsEnabled,
|
||||
|
|
|
|||
|
|
@ -266,6 +266,7 @@ function buildSkillPrimedIdsByName(manualSkillPrimes, alwaysApplySkillPrimes) {
|
|||
* @param {object} params.agent
|
||||
* @param {object} params.config
|
||||
* @param {Record<string, import('@librechat/api').LCAvailableTools>} [params.config.mcpAvailableTools]
|
||||
* @param {import('@librechat/api').RequestScopedMCPConnectionStore} [params.config.requestScopedConnections]
|
||||
* @returns {object}
|
||||
*/
|
||||
function buildAgentToolContext({ agent, config }) {
|
||||
|
|
@ -273,6 +274,7 @@ function buildAgentToolContext({ agent, config }) {
|
|||
agent,
|
||||
toolRegistry: config.toolRegistry,
|
||||
mcpAvailableTools: config.mcpAvailableTools,
|
||||
requestScopedConnections: config.requestScopedConnections,
|
||||
userMCPAuthMap: config.userMCPAuthMap,
|
||||
tool_resources: config.tool_resources,
|
||||
actionsEnabled: config.actionsEnabled,
|
||||
|
|
|
|||
|
|
@ -387,6 +387,7 @@ function createOAuthCallback({ runStepEmitter, runStepDeltaEmitter }) {
|
|||
* @param {number} [params.index]
|
||||
* @param {string | null} [params.streamId] - The stream ID for resumable mode.
|
||||
* @param {Record<string, Record<string, string>>} [params.userMCPAuthMap]
|
||||
* @param {import('@librechat/api').RequestScopedMCPConnectionStore} [params.requestScopedConnections]
|
||||
* @param {import('@librechat/api').ParsedServerConfig} [params.serverConfig] - Used to bypass reconnect throttling for request-scoped servers.
|
||||
* @returns { Promise<Array<typeof tool | { _call: (toolInput: Object | string) => unknown}>> } An object with `_call` method to execute the tool input.
|
||||
*/
|
||||
|
|
@ -400,6 +401,7 @@ async function reconnectServer({
|
|||
configServers,
|
||||
userMCPAuthMap,
|
||||
requestBody,
|
||||
requestScopedConnections,
|
||||
streamId = null,
|
||||
}) {
|
||||
logger.debug(
|
||||
|
|
@ -477,6 +479,7 @@ async function reconnectServer({
|
|||
flowManager,
|
||||
userMCPAuthMap,
|
||||
requestBody,
|
||||
requestScopedConnections,
|
||||
forceNew: true,
|
||||
returnOnOAuth: false,
|
||||
connectionTimeout: Time.THIRTY_SECONDS,
|
||||
|
|
@ -507,6 +510,7 @@ async function reconnectServer({
|
|||
* @param {string | null} [params.streamId] - The stream ID for resumable mode.
|
||||
* @param {import('@librechat/api').ParsedServerConfig} [params.config]
|
||||
* @param {import('@librechat/api').RequestBody} [params.requestBody]
|
||||
* @param {import('@librechat/api').RequestScopedMCPConnectionStore} [params.requestScopedConnections]
|
||||
* @param {Record<string, Record<string, string>>} [params.userMCPAuthMap]
|
||||
* @returns { Promise<Array<typeof tool | { _call: (toolInput: Object | string) => unknown}>> } An object with `_call` method to execute the tool input.
|
||||
*/
|
||||
|
|
@ -522,6 +526,7 @@ async function createMCPTools({
|
|||
configServers,
|
||||
userMCPAuthMap,
|
||||
requestBody,
|
||||
requestScopedConnections,
|
||||
streamId = null,
|
||||
}) {
|
||||
const serverConfig =
|
||||
|
|
@ -560,6 +565,7 @@ async function createMCPTools({
|
|||
configServers,
|
||||
userMCPAuthMap,
|
||||
requestBody,
|
||||
requestScopedConnections,
|
||||
streamId,
|
||||
});
|
||||
if (result === null) {
|
||||
|
|
@ -584,6 +590,7 @@ async function createMCPTools({
|
|||
availableTools: result.availableTools,
|
||||
toolKey: `${tool.name}${Constants.mcp_delimiter}${serverName}`,
|
||||
requestBody,
|
||||
requestScopedConnections,
|
||||
config: serverConfig,
|
||||
});
|
||||
if (toolInstance) {
|
||||
|
|
@ -608,6 +615,7 @@ async function createMCPTools({
|
|||
* @param {Providers | EModelEndpoint} params.provider - The provider for the tool.
|
||||
* @param {LCAvailableTools} [params.availableTools]
|
||||
* @param {import('@librechat/api').RequestBody} [params.requestBody]
|
||||
* @param {import('@librechat/api').RequestScopedMCPConnectionStore} [params.requestScopedConnections]
|
||||
* @param {Record<string, Record<string, string>>} [params.userMCPAuthMap]
|
||||
* @param {import('@librechat/api').ParsedServerConfig} [params.config]
|
||||
* @param {(availableTools: LCAvailableTools) => void} [params.onAvailableTools]
|
||||
|
|
@ -624,6 +632,7 @@ async function createMCPTool({
|
|||
userMCPAuthMap,
|
||||
availableTools,
|
||||
requestBody,
|
||||
requestScopedConnections,
|
||||
config,
|
||||
configServers,
|
||||
onAvailableTools,
|
||||
|
|
@ -683,6 +692,7 @@ async function createMCPTool({
|
|||
configServers,
|
||||
userMCPAuthMap,
|
||||
requestBody,
|
||||
requestScopedConnections,
|
||||
streamId,
|
||||
});
|
||||
if (result?.availableTools) {
|
||||
|
|
@ -708,6 +718,7 @@ async function createMCPTool({
|
|||
mcpPermissionContext,
|
||||
user,
|
||||
requestBody,
|
||||
requestScopedConnections,
|
||||
provider,
|
||||
toolName,
|
||||
serverName,
|
||||
|
|
@ -722,6 +733,7 @@ function createToolInstance({
|
|||
mcpPermissionContext,
|
||||
user: capturedUser = null,
|
||||
requestBody: capturedRequestBody,
|
||||
requestScopedConnections: capturedRequestScopedConnections,
|
||||
toolName,
|
||||
serverName,
|
||||
serverConfig: capturedServerConfig,
|
||||
|
|
@ -816,6 +828,8 @@ function createToolInstance({
|
|||
},
|
||||
user: effectiveUser,
|
||||
requestBody: config?.configurable?.requestBody ?? capturedRequestBody,
|
||||
requestScopedConnections:
|
||||
config?.configurable?.requestScopedConnections ?? capturedRequestScopedConnections,
|
||||
customUserVars,
|
||||
flowManager,
|
||||
tokenMethods: {
|
||||
|
|
|
|||
69
api/server/services/MCPRequestContext.js
Normal file
69
api/server/services/MCPRequestContext.js
Normal file
|
|
@ -0,0 +1,69 @@
|
|||
const { logger } = require('@librechat/data-schemas');
|
||||
|
||||
const MCP_REQUEST_CONTEXT = Symbol.for('librechat.mcpRequestContext');
|
||||
|
||||
function createMCPRequestContext() {
|
||||
return {
|
||||
connections: new Map(),
|
||||
pending: new Map(),
|
||||
cleanupStarted: false,
|
||||
};
|
||||
}
|
||||
|
||||
async function cleanupMCPRequestContext(context) {
|
||||
if (!context || context.cleanupStarted) {
|
||||
return;
|
||||
}
|
||||
context.cleanupStarted = true;
|
||||
|
||||
const connections = new Set(context.connections.values());
|
||||
const pending = Array.from(context.pending.values());
|
||||
if (pending.length > 0) {
|
||||
const settled = await Promise.allSettled(pending);
|
||||
for (const result of settled) {
|
||||
if (result.status === 'fulfilled' && result.value) {
|
||||
connections.add(result.value);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
await Promise.allSettled(
|
||||
Array.from(connections).map(async (connection) => {
|
||||
try {
|
||||
await connection.disconnect();
|
||||
} catch (error) {
|
||||
logger.warn('[MCP Request Context] Failed to disconnect request-scoped connection', error);
|
||||
}
|
||||
}),
|
||||
);
|
||||
|
||||
context.connections.clear();
|
||||
context.pending.clear();
|
||||
}
|
||||
|
||||
function getMCPRequestContext(req, res) {
|
||||
if (!req) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
if (!req[MCP_REQUEST_CONTEXT]) {
|
||||
const context = createMCPRequestContext();
|
||||
req[MCP_REQUEST_CONTEXT] = context;
|
||||
|
||||
const cleanup = () => {
|
||||
cleanupMCPRequestContext(context).catch((error) => {
|
||||
logger.warn('[MCP Request Context] Cleanup failed', error);
|
||||
});
|
||||
};
|
||||
res?.once?.('finish', cleanup);
|
||||
res?.once?.('close', cleanup);
|
||||
}
|
||||
|
||||
return req[MCP_REQUEST_CONTEXT];
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
cleanupMCPRequestContext,
|
||||
createMCPRequestContext,
|
||||
getMCPRequestContext,
|
||||
};
|
||||
|
|
@ -70,6 +70,7 @@ const { manifestToolMap, toolkits } = require('~/app/clients/tools/manifest');
|
|||
const { createOnSearchResults } = require('~/server/services/Tools/search');
|
||||
const { reinitMCPServer } = require('~/server/services/Tools/mcp');
|
||||
const { createMCPPermissionContext, resolveConfigServers } = require('~/server/services/MCP');
|
||||
const { getMCPRequestContext } = require('~/server/services/MCPRequestContext');
|
||||
const { recordUsage } = require('~/server/services/Threads');
|
||||
const { loadTools } = require('~/app/clients/tools/util');
|
||||
const { redactMessage } = require('~/config/parsers');
|
||||
|
|
@ -608,6 +609,7 @@ async function loadToolDefinitionsWrapper({ req, res, agent, streamId = null, to
|
|||
const oauthStepIndexes = new Map();
|
||||
/** @type {Record<string, import('@librechat/api').LCAvailableTools>} */
|
||||
const mcpAvailableTools = {};
|
||||
const requestScopedConnections = getMCPRequestContext(req, res);
|
||||
const rememberMCPAvailableTools = (serverName, availableTools) => {
|
||||
if (!availableTools || Object.keys(availableTools).length === 0) {
|
||||
return;
|
||||
|
|
@ -784,6 +786,7 @@ async function loadToolDefinitionsWrapper({ req, res, agent, streamId = null, to
|
|||
configServers,
|
||||
userMCPAuthMap,
|
||||
requestBody: req.body,
|
||||
requestScopedConnections,
|
||||
});
|
||||
|
||||
rememberMCPAvailableTools(serverName, result?.availableTools);
|
||||
|
|
@ -1041,6 +1044,7 @@ async function loadToolDefinitionsWrapper({ req, res, agent, streamId = null, to
|
|||
return {
|
||||
toolRegistry,
|
||||
mcpAvailableTools,
|
||||
requestScopedConnections,
|
||||
userMCPAuthMap,
|
||||
toolContextMap,
|
||||
dynamicToolContextMap,
|
||||
|
|
@ -1167,6 +1171,7 @@ async function loadAgentTools({
|
|||
uploadImageBuffer,
|
||||
returnMetadata: true,
|
||||
mcpPermissionContext,
|
||||
requestScopedConnections: getMCPRequestContext(req, res),
|
||||
[Tools.web_search]: webSearchCallbacks,
|
||||
},
|
||||
webSearch: appConfig.webSearch,
|
||||
|
|
@ -1242,6 +1247,7 @@ async function loadAgentTools({
|
|||
if (!hasActionTools) {
|
||||
return {
|
||||
toolRegistry,
|
||||
requestScopedConnections: getMCPRequestContext(req, res),
|
||||
userMCPAuthMap,
|
||||
toolContextMap,
|
||||
dynamicToolContextMap,
|
||||
|
|
@ -1260,6 +1266,7 @@ async function loadAgentTools({
|
|||
}
|
||||
return {
|
||||
toolRegistry,
|
||||
requestScopedConnections: getMCPRequestContext(req, res),
|
||||
userMCPAuthMap,
|
||||
toolContextMap,
|
||||
dynamicToolContextMap,
|
||||
|
|
@ -1388,6 +1395,7 @@ async function loadAgentTools({
|
|||
|
||||
return {
|
||||
toolRegistry,
|
||||
requestScopedConnections: getMCPRequestContext(req, res),
|
||||
toolContextMap,
|
||||
dynamicToolContextMap,
|
||||
userMCPAuthMap,
|
||||
|
|
@ -1414,6 +1422,7 @@ async function loadAgentTools({
|
|||
* @param {string[]} params.toolNames - Names of tools to load
|
||||
* @param {Map} [params.toolRegistry] - Tool registry
|
||||
* @param {Record<string, import('@librechat/api').LCAvailableTools>} [params.mcpAvailableTools] - Run-scoped MCP tool definitions
|
||||
* @param {import('@librechat/api').RequestScopedMCPConnectionStore} [params.requestScopedConnections] - Run-scoped MCP connections
|
||||
* @param {Record<string, Record<string, string>>} [params.userMCPAuthMap] - User MCP auth map
|
||||
* @param {Object} [params.tool_resources] - Tool resources
|
||||
* @param {string|null} [params.streamId] - Stream ID for web search callbacks
|
||||
|
|
@ -1428,6 +1437,7 @@ async function loadToolsForExecution({
|
|||
toolNames,
|
||||
toolRegistry,
|
||||
mcpAvailableTools,
|
||||
requestScopedConnections,
|
||||
userMCPAuthMap,
|
||||
tool_resources,
|
||||
streamId = null,
|
||||
|
|
@ -1435,7 +1445,8 @@ async function loadToolsForExecution({
|
|||
}) {
|
||||
const appConfig = req.config;
|
||||
const allLoadedTools = [];
|
||||
const configurable = { userMCPAuthMap };
|
||||
const mcpRequestScopedConnections = requestScopedConnections ?? getMCPRequestContext(req, res);
|
||||
const configurable = { userMCPAuthMap, requestScopedConnections: mcpRequestScopedConnections };
|
||||
|
||||
const isToolSearch = toolNames.includes(AgentConstants.TOOL_SEARCH);
|
||||
const ptcToolNames = [
|
||||
|
|
@ -1556,6 +1567,7 @@ async function loadToolsForExecution({
|
|||
uploadImageBuffer,
|
||||
returnMetadata: true,
|
||||
mcpAvailableTools,
|
||||
requestScopedConnections: mcpRequestScopedConnections,
|
||||
[Tools.web_search]: webSearchCallbacks,
|
||||
},
|
||||
webSearch: appConfig?.webSearch,
|
||||
|
|
|
|||
|
|
@ -24,6 +24,7 @@ const { getLogStores } = require('~/cache');
|
|||
* @param {(authURL: string, options?: { expiresAt?: number }) => Promise<void>} [params.oauthStart]
|
||||
* @param {() => Promise<void>} [params.oauthEnd]
|
||||
* @param {import('@librechat/api').RequestBody} [params.requestBody]
|
||||
* @param {import('@librechat/api').RequestScopedMCPConnectionStore} [params.requestScopedConnections]
|
||||
* @param {Record<string, Record<string, string>>} [params.userMCPAuthMap]
|
||||
*/
|
||||
async function reinitMCPServer({
|
||||
|
|
@ -39,6 +40,7 @@ async function reinitMCPServer({
|
|||
flowManager: _flowManager,
|
||||
serverConfig: providedConfig,
|
||||
requestBody,
|
||||
requestScopedConnections,
|
||||
oauthEnd,
|
||||
}) {
|
||||
/** @type {MCPConnection | null} */
|
||||
|
|
@ -144,6 +146,7 @@ async function reinitMCPServer({
|
|||
oauthEnd,
|
||||
customUserVars,
|
||||
requestBody,
|
||||
requestScopedConnections,
|
||||
connectionTimeout,
|
||||
serverConfig,
|
||||
graphTokenResolver: getGraphApiToken,
|
||||
|
|
@ -265,7 +268,7 @@ async function reinitMCPServer({
|
|||
error,
|
||||
);
|
||||
} finally {
|
||||
if (connection && ephemeralServer) {
|
||||
if (connection && ephemeralServer && !requestScopedConnections) {
|
||||
try {
|
||||
await connection.disconnect();
|
||||
} catch (error) {
|
||||
|
|
|
|||
|
|
@ -778,7 +778,7 @@ describe('ToolService - Action Capability Gating', () => {
|
|||
mockGetEndpointsConfig.mockResolvedValue(createEndpointsConfig(capabilities));
|
||||
mockGetServerConfig.mockResolvedValue({
|
||||
type: 'streamable-http',
|
||||
url: 'https://mcp.example.com/{{LIBRECHAT_OPENID_ACCESS_TOKEN}}/mcp',
|
||||
url: 'https://mcp.example.com/{{LIBRECHAT_BODY_MESSAGEID}}/mcp',
|
||||
source: 'yaml',
|
||||
});
|
||||
mockGetMCPServerTools.mockResolvedValue(null);
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue