diff --git a/libs/sdk/src/transport/__tests__/transport.registry.spec.ts b/libs/sdk/src/transport/__tests__/transport.registry.spec.ts index 567d79f90..8f49d7b43 100644 --- a/libs/sdk/src/transport/__tests__/transport.registry.spec.ts +++ b/libs/sdk/src/transport/__tests__/transport.registry.spec.ts @@ -6,6 +6,7 @@ import { sha256Hex } from '@frontmcp/utils'; import type { TransportPersistenceConfigInput } from '../../common'; +import { loadSseTransport } from '../adapters/transport.sse.adapter'; import { loadStreamableHttpTransport } from '../adapters/transport.streamable-http.adapter'; import { TransportService } from '../transport.registry'; @@ -32,6 +33,11 @@ jest.mock('@frontmcp/auth', () => { }; }); +jest.mock('../adapters/transport.sse.adapter', () => ({ + ...jest.requireActual('../adapters/transport.sse.adapter'), + loadSseTransport: jest.fn().mockResolvedValue(undefined), +})); + jest.mock('../adapters/transport.streamable-http.adapter', () => ({ ...jest.requireActual('../adapters/transport.streamable-http.adapter'), loadStreamableHttpTransport: jest.fn().mockResolvedValue(undefined), @@ -350,6 +356,15 @@ describe('TransportService', () => { expect(loadStreamableHttpTransport).toHaveBeenCalledTimes(2); }); + it('loads the SSE transport only for SSE sessions', async () => { + await service.createTransporter('streamable-http', 'test-token', 'session-http', mockResponse as never); + await service.createTransporter('stateless-http', 'test-token', 'session-stateless', mockResponse as never); + expect(loadSseTransport).not.toHaveBeenCalled(); + + await service.createTransporter('sse', 'test-token', 'session-sse', mockResponse as never); + expect(loadSseTransport).toHaveBeenCalledTimes(1); + }); + it('should persist session to Redis for streamable-http', async () => { await service.createTransporter('streamable-http', 'test-token', 'session-123', mockResponse as never); expect(mockRedisSessionStore.set).toHaveBeenCalled(); diff --git a/libs/sdk/src/transport/adapters/__tests__/transport.sse.adapter.load.spec.ts b/libs/sdk/src/transport/adapters/__tests__/transport.sse.adapter.load.spec.ts new file mode 100644 index 000000000..817cab76d --- /dev/null +++ b/libs/sdk/src/transport/adapters/__tests__/transport.sse.adapter.load.spec.ts @@ -0,0 +1,41 @@ +import 'reflect-metadata'; + +import { InternalMcpError } from '../../../errors'; +import { RecreateableSSEServerTransport } from '../sse-transport'; +import { loadSseTransport, TransportSSEAdapter } from '../transport.sse.adapter'; + +jest.mock('../sse-transport', () => ({ + RecreateableSSEServerTransport: jest.fn().mockImplementation(() => ({})), +})); + +function makeAdapter(): TransportSSEAdapter { + const adapter = Object.create(TransportSSEAdapter.prototype) as TransportSSEAdapter; + Object.assign(adapter, { + logger: { info: jest.fn(), verbose: jest.fn() }, + scope: { fullPath: '/mcp' }, + }); + return adapter; +} + +describe('TransportSSEAdapter loads its transport class on demand', () => { + it('refuses to create a transport before loadSseTransport() loaded the class', () => { + expect(() => makeAdapter().createTransport('session-1', {} as never)).toThrow(InternalMcpError); + expect(RecreateableSSEServerTransport).not.toHaveBeenCalled(); + }); + + it('creates and recreates transports with the loaded class', async () => { + await loadSseTransport(); + const response = {} as never; + + makeAdapter().createTransport('session-1', response); + makeAdapter().createTransportFromSession('session-2', response, 41); + + expect(RecreateableSSEServerTransport).toHaveBeenNthCalledWith(1, '/mcp/message', response, { + sessionId: 'session-1', + }); + expect(RecreateableSSEServerTransport).toHaveBeenNthCalledWith(2, '/mcp/message', response, { + sessionId: 'session-2', + initialEventId: 41, + }); + }); +}); diff --git a/libs/sdk/src/transport/adapters/transport.sse.adapter.ts b/libs/sdk/src/transport/adapters/transport.sse.adapter.ts index a35a6aa19..5c91fa8a8 100644 --- a/libs/sdk/src/transport/adapters/transport.sse.adapter.ts +++ b/libs/sdk/src/transport/adapters/transport.sse.adapter.ts @@ -3,12 +3,26 @@ import { type RequestId } from '@frontmcp/protocol'; import { type ServerResponse } from '../../common'; import { DEFAULT_ELICIT_TTL, type ElicitOptions, type ElicitResult } from '../../elicitation/elicitation.types'; -import { ElicitationTimeoutError, InvalidInputError } from '../../errors'; +import { ElicitationTimeoutError, InternalMcpError, InvalidInputError } from '../../errors'; import { type AuthenticatedServerRequest } from '../../server/server.types'; import { rpcRequest } from '../transport.error'; -import { RecreateableSSEServerTransport } from './sse-transport'; +import type { RecreateableSSEServerTransport } from './sse-transport'; import { LocalTransportAdapter } from './transport.local.adapter'; +let RecreateableTransport: typeof RecreateableSSEServerTransport | undefined; + +// The SSE transport reads POST bodies with raw-body and content-type, which a stdio server never needs. +export async function loadSseTransport(): Promise { + RecreateableTransport ??= (await import('./sse-transport.js')).RecreateableSSEServerTransport; +} + +function sseTransportClass(): typeof RecreateableSSEServerTransport { + if (!RecreateableTransport) { + throw new InternalMcpError('The SSE transport was used before loadSseTransport() loaded it'); + } + return RecreateableTransport; +} + export class TransportSSEAdapter extends LocalTransportAdapter { sessionId: string; @@ -26,7 +40,8 @@ export class TransportSSEAdapter extends LocalTransportAdapter> = new Map(); - // The Streamable HTTP transport class loads with the first session that uses it, so stdio and SSE servers never load it. + // Each transport class loads with the first session that uses it, so a stdio server loads neither. private async newLocalTransporter( key: TransportKey, res: ServerResponse, @@ -123,6 +124,8 @@ export class TransportService { ): Promise { if (key.type === 'streamable-http' || key.type === 'stateless-http') { await loadStreamableHttpTransport(); + } else if (key.type === 'sse') { + await loadSseTransport(); } return new LocalTransporter(this.scope, key, res, onDispose); }