diff --git a/apps/api/src/mcp-clients/outbound/index.ts b/apps/api/src/mcp-clients/outbound/index.ts index 912f2e91ba..7e613f2401 100644 --- a/apps/api/src/mcp-clients/outbound/index.ts +++ b/apps/api/src/mcp-clients/outbound/index.ts @@ -28,6 +28,45 @@ import { // Separate from the per-request pool on StudioContext (used by HTTP/SSE). const stdioPool = createClientPool(); +// Shared by HTTP/Websocket and SSE — only the transport class differs. +async function buildHttpLikeTransport( + connection: ConnectionEntity, + ctx: StudioContext, + superUser: boolean, + virtualMcpId: string | undefined, + TransportCtor: new ( + url: URL, + opts: { requestInit: { headers: Record } }, + ) => Transport, +): Promise { + if (!connection.connection_url) { + throw new Error(`${connection.connection_type} connection missing URL`); + } + + const headers = await buildRequestHeaders(connection, ctx, superUser); + + const httpParams = connection.connection_headers; + if (httpParams && "headers" in httpParams) { + Object.assign(headers, httpParams.headers); + } + + const transport: Transport = new TransportCtor( + new URL(connection.connection_url), + { requestInit: { headers } }, + ); + + return composeTransport( + transport, + (t) => new AuthTransport(t, { ctx, connection, superUser }), + (t) => + new MonitoringTransport(t, { + ctx, + connectionId: connection.id, + virtualMcpId, + }), + ); +} + /** * Create an MCP client for outbound connections (STDIO, HTTP, Websocket, SSE) * @@ -94,66 +133,24 @@ export async function createOutboundClient( case "HTTP": case "Websocket": { - if (!connection.connection_url) { - throw new Error(`${connection.connection_type} connection missing URL`); - } - - const headers = await buildRequestHeaders(connection, ctx, superUser); - - const httpParams = connection.connection_headers; - if (httpParams && "headers" in httpParams) { - Object.assign(headers, httpParams.headers); - } - - let transport: Transport = new StreamableHTTPClientTransport( - new URL(connection.connection_url), - { requestInit: { headers } }, + const transport = await buildHttpLikeTransport( + connection, + ctx, + superUser, + virtualMcpId, + StreamableHTTPClientTransport, ); - - // Compose with auth and monitoring transports - transport = composeTransport( - transport, - (t) => new AuthTransport(t, { ctx, connection, superUser }), - (t) => - new MonitoringTransport(t, { - ctx, - connectionId, - virtualMcpId, - }), - ); - return ctx.getOrCreateClient(transport, connectionId); } case "SSE": { - if (!connection.connection_url) { - throw new Error("SSE connection missing URL"); - } - - const headers = await buildRequestHeaders(connection, ctx, superUser); - - const httpParams = connection.connection_headers; - if (httpParams && "headers" in httpParams) { - Object.assign(headers, httpParams.headers); - } - - let transport: Transport = new SSEClientTransport( - new URL(connection.connection_url), - { requestInit: { headers } }, - ); - - // Compose with auth and monitoring transports - transport = composeTransport( - transport, - (t) => new AuthTransport(t, { ctx, connection, superUser }), - (t) => - new MonitoringTransport(t, { - ctx, - connectionId, - virtualMcpId, - }), + const transport = await buildHttpLikeTransport( + connection, + ctx, + superUser, + virtualMcpId, + SSEClientTransport, ); - return ctx.getOrCreateClient(transport, connectionId); }