Add SSE streaming support on POST /mcp (Phase 2) (#19528)

## Summary - Add SSE (`text/event-stream`) as an alternative response format on `POST /mcp` per the MCP streamable-http spec - When clients send `Accept: text/event-stream`, the server responds with SSE wire format; otherwise returns JSON as before (fully backwards compatible) - Emit a `notifications/progress` SSE event before tool execution to signal long-running operations - No sessions, no GET SSE, no protocol version bump — this is transport-level only (Phase 2) ## Changes - **New**: `write-sse-event.util.ts` — writes correctly formatted SSE events to Express Response - **New**: `mcp-progress-notification.const.ts` — constants for progress notification method and token prefix - **Modified**: `mcp-core.controller.ts` — checks `Accept` header, branches into SSE vs JSON response path - **Modified**: `mcp-protocol.service.ts` — passes optional `sseWriter` callback to tool executor - **Modified**: `mcp-tool-executor.service.ts` — emits progress notification via `sseWriter` before tool execution - **Tests**: Unit tests for SSE utility, controller SSE/JSON paths, tool executor progress notifications, and integration tests for SSE streaming ## Test plan - [x] Unit tests: `writeSseEvent` utility produces correct SSE wire format - [x] Unit tests: Controller returns SSE headers and writes events when `Accept: text/event-stream` - [x] Unit tests: Controller returns JSON when `Accept: application/json` only - [x] Unit tests: Notifications (no `id`) return 202 regardless of Accept header - [x] Unit tests: Tool executor emits progress notification via sseWriter - [x] Unit tests: Tool executor works without sseWriter (backwards compatible) - [x] Integration tests: SSE response for ping, JSON fallback, progress notification before tool call - [x] All 503 test suites pass, typecheck clean https://claude.ai/code/session_01QrqjBUXePJkPMd6gBAoWaR --------- Co-authored-by: Claude <noreply@anthropic.com>
2026-04-10 09:56:17 +02:00
parent f6423f5925
commit cbefd42c4c
10 changed files with 450 additions and 5 deletions
@@ -0,0 +1,3 @@
+export const MCP_PROGRESS_NOTIFICATION_METHOD = 'notifications/progress';
+
+export const TOOL_CALL_PROGRESS_TOKEN_PREFIX = 'tool-call-';
@@ -75,10 +75,16 @@ describe('McpCoreController', () => {
    const mockApiKey = { id: 'api-key-1' } as FlatApiKey;
    const mockRes = {
      status: jest.fn().mockReturnThis(),
+      setHeader: jest.fn(),
+      write: jest.fn(),
+      end: jest.fn(),
    } as unknown as import('express').Response;

    beforeEach(() => {
      (mockRes.status as jest.Mock).mockClear();
+      (mockRes.setHeader as jest.Mock).mockClear();
+      (mockRes.write as jest.Mock).mockClear();
+      (mockRes.end as jest.Mock).mockClear();
    });

    it('should call mcpProtocolService.handleMCPCoreQuery with correct parameters', async () => {
@@ -106,6 +112,7 @@ describe('McpCoreController', () => {
        mockApiKey,
        mockUser,
        mockUserWorkspaceId,
+        undefined,
        mockRes,
      );

@@ -151,6 +158,7 @@ describe('McpCoreController', () => {
        mockApiKey,
        mockUser,
        mockUserWorkspaceId,
+        undefined,
        mockRes,
      );

@@ -195,6 +203,7 @@ describe('McpCoreController', () => {
        mockApiKey,
        mockUser,
        mockUserWorkspaceId,
+        undefined,
        mockRes,
      );

@@ -224,6 +233,7 @@ describe('McpCoreController', () => {
        mockApiKey,
        mockUser,
        mockUserWorkspaceId,
+        undefined,
        mockRes,
      );

@@ -256,6 +266,7 @@ describe('McpCoreController', () => {
        mockApiKey,
        undefined,
        undefined,
+        undefined,
        mockRes,
      );

@@ -270,5 +281,124 @@ describe('McpCoreController', () => {
      );
      expect(result).toEqual(mockResponse);
    });
+
+    it('should respond with SSE when client accepts text/event-stream', async () => {
+      const mockRequest: JsonRpc = {
+        jsonrpc: '2.0',
+        method: 'tools/call',
+        params: { name: 'testTool', arguments: {} },
+        id: '789',
+      };
+
+      const mockResponse = {
+        id: '789',
+        jsonrpc: '2.0',
+        result: {
+          content: [{ type: 'text', text: '{"ok":true}' }],
+          isError: false,
+        },
+      };
+
+      mcpProtocolService.handleMCPCoreQuery.mockResolvedValue(mockResponse);
+
+      const result = await controller.handleMcpCore(
+        mockRequest,
+        mockWorkspace,
+        mockApiKey,
+        mockUser,
+        mockUserWorkspaceId,
+        'application/json, text/event-stream',
+        mockRes,
+      );
+
+      // SSE path returns nothing — response is written directly
+      expect(result).toBeUndefined();
+      expect(mockRes.setHeader).toHaveBeenCalledWith(
+        'Content-Type',
+        'text/event-stream',
+      );
+      expect(mockRes.setHeader).toHaveBeenCalledWith(
+        'Cache-Control',
+        'no-cache',
+      );
+      expect(mockRes.setHeader).toHaveBeenCalledWith(
+        'Connection',
+        'keep-alive',
+      );
+      expect(mockRes.setHeader).toHaveBeenCalledWith(
+        'X-Content-Type-Options',
+        'nosniff',
+      );
+      expect(mockRes.write).toHaveBeenCalledWith(
+        `event: message\ndata: ${JSON.stringify(mockResponse)}\n\n`,
+      );
+      expect(mockRes.end).toHaveBeenCalled();
+      // sseWriter callback should be passed to protocol service
+      expect(mcpProtocolService.handleMCPCoreQuery).toHaveBeenCalledWith(
+        mockRequest,
+        {
+          workspace: mockWorkspace,
+          userId: mockUser.id,
+          userWorkspaceId: mockUserWorkspaceId,
+          apiKey: mockApiKey,
+        },
+        expect.any(Function),
+      );
+    });
+
+    it('should return JSON when client only accepts application/json', async () => {
+      const mockRequest: JsonRpc = {
+        jsonrpc: '2.0',
+        method: 'ping',
+        id: '101',
+      };
+
+      const mockResponse = {
+        id: '101',
+        jsonrpc: '2.0',
+        result: {},
+      };
+
+      mcpProtocolService.handleMCPCoreQuery.mockResolvedValue(mockResponse);
+
+      const result = await controller.handleMcpCore(
+        mockRequest,
+        mockWorkspace,
+        mockApiKey,
+        mockUser,
+        mockUserWorkspaceId,
+        'application/json',
+        mockRes,
+      );
+
+      expect(result).toEqual(mockResponse);
+      expect(mockRes.setHeader).not.toHaveBeenCalled();
+      expect(mockRes.write).not.toHaveBeenCalled();
+      expect(mockRes.end).not.toHaveBeenCalled();
+    });
+
+    it('should return 202 for notifications even when client accepts SSE', async () => {
+      const mockRequest: JsonRpc = {
+        jsonrpc: '2.0',
+        method: 'notifications/initialized',
+      };
+
+      mcpProtocolService.handleMCPCoreQuery.mockResolvedValue(null);
+
+      const result = await controller.handleMcpCore(
+        mockRequest,
+        mockWorkspace,
+        mockApiKey,
+        mockUser,
+        mockUserWorkspaceId,
+        'application/json, text/event-stream',
+        mockRes,
+      );
+
+      expect(result).toBeUndefined();
+      expect(mockRes.status).toHaveBeenCalledWith(202);
+      expect(mockRes.setHeader).not.toHaveBeenCalled();
+      expect(mockRes.end).not.toHaveBeenCalled();
+    });
  });
 });
@@ -1,6 +1,7 @@
 import {
  Body,
  Controller,
+  Headers,
  HttpCode,
  HttpStatus,
  Post,
@@ -17,6 +18,7 @@ import { isDefined } from 'twenty-shared/utils';
 import { JsonRpc } from 'src/engine/api/mcp/dtos/json-rpc';
 import { McpAuthGuard } from 'src/engine/api/mcp/guards/mcp-auth.guard';
 import { McpProtocolService } from 'src/engine/api/mcp/services/mcp-protocol.service';
+import { writeSseEvent } from 'src/engine/api/mcp/utils/write-sse-event.util';
 import { RestApiExceptionFilter } from 'src/engine/api/rest/rest-api-exception.filter';
 import { FlatApiKey } from 'src/engine/core-modules/api-key/types/flat-api-key.type';
 import { FlatWorkspace } from 'src/engine/core-modules/workspace/types/flat-workspace.type';
@@ -50,22 +52,62 @@ export class McpCoreController {
    @AuthUser({ allowUndefined: true }) user: UserEntity | undefined,
    @AuthUserWorkspaceId({ allowUndefined: true })
    userWorkspaceId: string | undefined,
+    @Headers('accept') acceptHeader: string | undefined,
    @Res({ passthrough: true }) res: Response,
  ) {
-    const result = await this.mcpProtocolService.handleMCPCoreQuery(body, {
+    const authContext = {
      workspace,
      userId: user?.id,
      userWorkspaceId,
      apiKey,
-    });
+    };
+
+    // JSON-RPC notifications (no id) expect no response body regardless of Accept
+    if (!isDefined(body.id)) {
+      await this.mcpProtocolService.handleMCPCoreQuery(body, authContext);

-    // JSON-RPC notifications (no id) expect no response body
-    if (!isDefined(result)) {
      res.status(HttpStatus.ACCEPTED);

      return;
    }

+    const clientAcceptsSse =
+      isDefined(acceptHeader) &&
+      acceptHeader
+        .split(',')
+        .some((type) => type.trim().startsWith('text/event-stream'));
+
+    if (clientAcceptsSse) {
+      res.setHeader('Content-Type', 'text/event-stream');
+      res.setHeader('Cache-Control', 'no-cache');
+      res.setHeader('Connection', 'keep-alive');
+      // Prevent browsers from MIME-sniffing the SSE stream as HTML
+      res.setHeader('X-Content-Type-Options', 'nosniff');
+
+      const sseWriter = (data: Record<string, unknown>) => {
+        writeSseEvent(res, data);
+      };
+
+      const result = await this.mcpProtocolService.handleMCPCoreQuery(
+        body,
+        authContext,
+        sseWriter,
+      );
+
+      if (isDefined(result)) {
+        writeSseEvent(res, result);
+      }
+
+      res.end();
+
+      return;
+    }
+
+    const result = await this.mcpProtocolService.handleMCPCoreQuery(
+      body,
+      authContext,
+    );
+
    return result;
  }
 }
@@ -261,6 +261,7 @@ describe('McpProtocolService', () => {
          ),
        ),
        mockRequest.params,
+        undefined,
      );
    });

@@ -1,4 +1,8 @@
 import { JSON_RPC_ERROR_CODE } from 'src/engine/api/mcp/constants/json-rpc-error-code.const';
+import {
+  MCP_PROGRESS_NOTIFICATION_METHOD,
+  TOOL_CALL_PROGRESS_TOKEN_PREFIX,
+} from 'src/engine/api/mcp/constants/mcp-progress-notification.const';
 import { McpToolExecutorService } from 'src/engine/api/mcp/services/mcp-tool-executor.service';

 describe('McpToolExecutorService', () => {
@@ -142,5 +146,59 @@ describe('McpToolExecutorService', () => {
        },
      });
    });
+
+    it('should emit progress notification via sseWriter before execution', async () => {
+      const sseWriter = jest.fn();
+      const toolSet = {
+        my_tool: {
+          execute: jest.fn().mockResolvedValue({ data: 'ok' }),
+          description: 'My tool',
+          inputSchema: { type: 'object' },
+        },
+      } as any;
+
+      await service.handleToolCall(
+        'sse-1',
+        toolSet,
+        { name: 'my_tool', arguments: {} },
+        sseWriter,
+      );
+
+      expect(sseWriter).toHaveBeenCalledTimes(1);
+      expect(sseWriter).toHaveBeenCalledWith({
+        jsonrpc: '2.0',
+        method: MCP_PROGRESS_NOTIFICATION_METHOD,
+        params: {
+          progressToken: `${TOOL_CALL_PROGRESS_TOKEN_PREFIX}sse-1`,
+          progress: 0,
+          total: 1,
+        },
+      });
+    });
+
+    it('should not emit progress notification when sseWriter is undefined', async () => {
+      const toolSet = {
+        my_tool: {
+          execute: jest.fn().mockResolvedValue({ data: 'ok' }),
+          description: 'My tool',
+          inputSchema: { type: 'object' },
+        },
+      } as any;
+
+      const result = await service.handleToolCall('no-sse', toolSet, {
+        name: 'my_tool',
+        arguments: {},
+      });
+
+      // Should still return normal result without errors
+      expect(result).toEqual({
+        id: 'no-sse',
+        jsonrpc: '2.0',
+        result: {
+          content: [{ type: 'text', text: '{"data":"ok"}' }],
+          isError: false,
+        },
+      });
+    });
  });
 });
@@ -178,6 +178,7 @@ export class McpProtocolService {
      userWorkspaceId?: string;
      apiKey: FlatApiKey | undefined;
    },
+    sseWriter?: (data: Record<string, unknown>) => void,
  ): Promise<Record<string, unknown> | null> {
    try {
      // JSON-RPC notifications have no id and expect no response
@@ -244,6 +245,7 @@ export class McpProtocolService {
          id,
          toolSet,
          params,
+          sseWriter,
        );
      }

@@ -4,6 +4,10 @@ import { type ToolSet } from 'ai';
 import { isDefined } from 'twenty-shared/utils';

 import { JSON_RPC_ERROR_CODE } from 'src/engine/api/mcp/constants/json-rpc-error-code.const';
+import {
+  MCP_PROGRESS_NOTIFICATION_METHOD,
+  TOOL_CALL_PROGRESS_TOKEN_PREFIX,
+} from 'src/engine/api/mcp/constants/mcp-progress-notification.const';
 import { wrapJsonRpcResponse } from 'src/engine/api/mcp/utils/wrap-jsonrpc-response.util';

@Injectable()
@@ -12,6 +16,7 @@ export class McpToolExecutorService {
    id: string | number,
    toolSet: ToolSet,
    params: Record<string, unknown>,
+    sseWriter?: (data: Record<string, unknown>) => void,
  ) {
    const toolName = params.name as keyof typeof toolSet;
    const tool = toolSet[toolName];
@@ -25,6 +30,18 @@ export class McpToolExecutorService {
      });
    }

+    if (isDefined(sseWriter)) {
+      sseWriter({
+        jsonrpc: '2.0',
+        method: MCP_PROGRESS_NOTIFICATION_METHOD,
+        params: {
+          progressToken: `${TOOL_CALL_PROGRESS_TOKEN_PREFIX}${String(id)}`,
+          progress: 0,
+          total: 1,
+        },
+      });
+    }
+
    try {
      const result = await tool.execute(params.arguments, {
        toolCallId: '1',
@@ -0,0 +1,67 @@
+import { writeSseEvent } from 'src/engine/api/mcp/utils/write-sse-event.util';
+
+const createMockResponse = (headersSent = false) => ({
+  write: jest.fn(),
+  setHeader: jest.fn(),
+  headersSent,
+});
+
+describe('writeSseEvent', () => {
+  it('should write correctly formatted SSE event', () => {
+    const mockResponse = createMockResponse() as any;
+    const data = { jsonrpc: '2.0', id: '123', result: { foo: 'bar' } };
+
+    writeSseEvent(mockResponse, data);
+
+    expect(mockResponse.write).toHaveBeenCalledTimes(1);
+    expect(mockResponse.write).toHaveBeenCalledWith(
+      `event: message\ndata: ${JSON.stringify(data)}\n\n`,
+    );
+  });
+
+  it('should produce wire format per HTML SSE spec', () => {
+    const mockResponse = createMockResponse() as any;
+    const data = { jsonrpc: '2.0', id: 1, result: {} };
+
+    writeSseEvent(mockResponse, data);
+
+    const written = mockResponse.write.mock.calls[0][0] as string;
+
+    // Must start with "event: message\n"
+    expect(written.startsWith('event: message\n')).toBe(true);
+    // Must contain "data: " followed by valid JSON
+    const dataLine = written.split('\n')[1];
+
+    expect(dataLine).toBeDefined();
+    expect(dataLine!.startsWith('data: ')).toBe(true);
+    const parsed = JSON.parse(dataLine!.slice('data: '.length));
+
+    expect(parsed).toEqual(data);
+    // Must end with double newline to terminate the event
+    expect(written.endsWith('\n\n')).toBe(true);
+  });
+
+  it('should set content-type headers when headers have not been sent', () => {
+    const mockResponse = createMockResponse(false) as any;
+
+    writeSseEvent(mockResponse, { jsonrpc: '2.0', id: '1', result: {} });
+
+    expect(mockResponse.setHeader).toHaveBeenCalledWith(
+      'Content-Type',
+      'text/event-stream',
+    );
+    expect(mockResponse.setHeader).toHaveBeenCalledWith(
+      'X-Content-Type-Options',
+      'nosniff',
+    );
+  });
+
+  it('should skip setting headers when headers have already been sent', () => {
+    const mockResponse = createMockResponse(true) as any;
+
+    writeSseEvent(mockResponse, { jsonrpc: '2.0', id: '1', result: {} });
+
+    expect(mockResponse.setHeader).not.toHaveBeenCalled();
+    expect(mockResponse.write).toHaveBeenCalledTimes(1);
+  });
+});
@@ -0,0 +1,14 @@
+import { type Response } from 'express';
+
+export const writeSseEvent = (
+  response: Response,
+  data: Record<string, unknown>,
+): void => {
+  // Ensure non-HTML content type so user data serialized as JSON cannot trigger XSS
+  if (!response.headersSent) {
+    response.setHeader('Content-Type', 'text/event-stream');
+    response.setHeader('X-Content-Type-Options', 'nosniff');
+  }
+
+  response.write(`event: message\ndata: ${JSON.stringify(data)}\n\n`);
+};
@@ -11,11 +11,16 @@ describe('MCP Controller (integration)', () => {
  const baseUrl = `http://localhost:${APP_PORT}`;
  const endpoint = '/mcp';

-  const postMcp = (body: any, bearer: string = API_KEY_ACCESS_TOKEN) => {
+  const postMcp = (
+    body: any,
+    bearer: string = API_KEY_ACCESS_TOKEN,
+    accept: string = 'application/json',
+  ) => {
    return request(baseUrl)
      .post(endpoint)
      .set('Authorization', `Bearer ${bearer}`)
      .set('Content-Type', 'application/json')
+      .set('Accept', accept)
      .send(JSON.stringify(body));
  };

@@ -191,4 +196,110 @@ describe('MCP Controller (integration)', () => {
      expect(Array.isArray(resources.body.result.resources)).toBe(true);
    });
  });
+
+  describe('SSE Streaming', () => {
+    const parseSseEvents = (
+      rawBody: string,
+    ): Array<Record<string, unknown>> => {
+      return rawBody
+        .split('\n\n')
+        .filter((block) => block.trim().length > 0)
+        .map((block) => {
+          const dataLine = block
+            .split('\n')
+            .find((line) => line.startsWith('data: '));
+
+          if (!dataLine) {
+            throw new Error(`No data line found in SSE block: ${block}`);
+          }
+
+          return JSON.parse(dataLine.slice('data: '.length));
+        });
+    };
+
+    it('should respond with SSE when Accept includes text/event-stream', async () => {
+      const res = await postMcp(
+        { jsonrpc: '2.0', method: 'ping', id: 'sse-ping-1' },
+        API_KEY_ACCESS_TOKEN,
+        'application/json, text/event-stream',
+      ).expect(200);
+
+      expect(res.headers['content-type']).toMatch(/text\/event-stream/);
+
+      const events = parseSseEvents(res.text);
+
+      expect(events.length).toBeGreaterThanOrEqual(1);
+
+      // The last event should be the JSON-RPC response
+      const lastEvent = events[events.length - 1];
+
+      expect(lastEvent).toMatchObject({
+        id: 'sse-ping-1',
+        jsonrpc: '2.0',
+        result: {},
+      });
+    });
+
+    it('should respond with JSON when Accept is application/json only', async () => {
+      const res = await postMcp(
+        { jsonrpc: '2.0', method: 'ping', id: 'json-ping-1' },
+        API_KEY_ACCESS_TOKEN,
+        'application/json',
+      ).expect(200);
+
+      expect(res.headers['content-type']).toMatch(/application\/json/);
+      expect(res.body).toMatchObject({
+        id: 'json-ping-1',
+        jsonrpc: '2.0',
+        result: {},
+      });
+    });
+
+    it('should include progress notification before tool call result in SSE', async () => {
+      const res = await postMcp(
+        {
+          jsonrpc: '2.0',
+          method: 'tools/call',
+          id: 'sse-tool-1',
+          params: { name: 'get_tool_catalog', arguments: {} },
+        },
+        API_KEY_ACCESS_TOKEN,
+        'application/json, text/event-stream',
+      ).expect(200);
+
+      expect(res.headers['content-type']).toMatch(/text\/event-stream/);
+
+      const events = parseSseEvents(res.text);
+
+      // Should have at least a progress notification and the final response
+      expect(events.length).toBeGreaterThanOrEqual(2);
+
+      // First event should be a progress notification
+      const progressEvent = events[0];
+
+      expect(progressEvent).toMatchObject({
+        jsonrpc: '2.0',
+        method: 'notifications/progress',
+        params: {
+          progressToken: 'tool-call-sse-tool-1',
+          progress: 0,
+          total: 1,
+        },
+      });
+
+      // Last event should be the final JSON-RPC response
+      const lastEvent = events[events.length - 1];
+
+      expect(lastEvent).toMatchObject({
+        id: 'sse-tool-1',
+        jsonrpc: '2.0',
+      });
+
+      // Should have either result.content or error
+      const hasResultContent = !!(lastEvent as any)?.result?.content;
+      const hasError = !!(lastEvent as any)?.error;
+
+      expect(hasResultContent || hasError).toBe(true);
+    });
+  });
 });