From cbefd42c4c9d777b394325523b16184a4eb97677 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?F=C3=A9lix=20Malfait?= Date: Fri, 10 Apr 2026 09:56:17 +0200 Subject: [PATCH] Add SSE streaming support on POST /mcp (Phase 2) (#19528) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 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 --- .../mcp-progress-notification.const.ts | 3 + .../__tests__/mcp-core.controller.spec.ts | 130 ++++++++++++++++++ .../mcp/controllers/mcp-core.controller.ts | 50 ++++++- .../__tests__/mcp-protocol.service.spec.ts | 1 + .../mcp-tool-executor.service.spec.ts | 58 ++++++++ .../api/mcp/services/mcp-protocol.service.ts | 2 + .../mcp/services/mcp-tool-executor.service.ts | 17 +++ .../__tests__/write-sse-event.util.spec.ts | 67 +++++++++ .../api/mcp/utils/write-sse-event.util.ts | 14 ++ .../suites/mcp.controller.integration-spec.ts | 113 ++++++++++++++- 10 files changed, 450 insertions(+), 5 deletions(-) create mode 100644 packages/twenty-server/src/engine/api/mcp/constants/mcp-progress-notification.const.ts create mode 100644 packages/twenty-server/src/engine/api/mcp/utils/__tests__/write-sse-event.util.spec.ts create mode 100644 packages/twenty-server/src/engine/api/mcp/utils/write-sse-event.util.ts diff --git a/packages/twenty-server/src/engine/api/mcp/constants/mcp-progress-notification.const.ts b/packages/twenty-server/src/engine/api/mcp/constants/mcp-progress-notification.const.ts new file mode 100644 index 0000000000..75b713ed69 --- /dev/null +++ b/packages/twenty-server/src/engine/api/mcp/constants/mcp-progress-notification.const.ts @@ -0,0 +1,3 @@ +export const MCP_PROGRESS_NOTIFICATION_METHOD = 'notifications/progress'; + +export const TOOL_CALL_PROGRESS_TOKEN_PREFIX = 'tool-call-'; diff --git a/packages/twenty-server/src/engine/api/mcp/controllers/__tests__/mcp-core.controller.spec.ts b/packages/twenty-server/src/engine/api/mcp/controllers/__tests__/mcp-core.controller.spec.ts index 5b986af6bd..966c4d278a 100644 --- a/packages/twenty-server/src/engine/api/mcp/controllers/__tests__/mcp-core.controller.spec.ts +++ b/packages/twenty-server/src/engine/api/mcp/controllers/__tests__/mcp-core.controller.spec.ts @@ -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(); + }); }); }); diff --git a/packages/twenty-server/src/engine/api/mcp/controllers/mcp-core.controller.ts b/packages/twenty-server/src/engine/api/mcp/controllers/mcp-core.controller.ts index e452ec6f5e..f24c484374 100644 --- a/packages/twenty-server/src/engine/api/mcp/controllers/mcp-core.controller.ts +++ b/packages/twenty-server/src/engine/api/mcp/controllers/mcp-core.controller.ts @@ -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) => { + 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; } } diff --git a/packages/twenty-server/src/engine/api/mcp/services/__tests__/mcp-protocol.service.spec.ts b/packages/twenty-server/src/engine/api/mcp/services/__tests__/mcp-protocol.service.spec.ts index 17853adb92..fc2e9c4199 100644 --- a/packages/twenty-server/src/engine/api/mcp/services/__tests__/mcp-protocol.service.spec.ts +++ b/packages/twenty-server/src/engine/api/mcp/services/__tests__/mcp-protocol.service.spec.ts @@ -261,6 +261,7 @@ describe('McpProtocolService', () => { ), ), mockRequest.params, + undefined, ); }); diff --git a/packages/twenty-server/src/engine/api/mcp/services/__tests__/mcp-tool-executor.service.spec.ts b/packages/twenty-server/src/engine/api/mcp/services/__tests__/mcp-tool-executor.service.spec.ts index b65fa4626e..ef6a13a392 100644 --- a/packages/twenty-server/src/engine/api/mcp/services/__tests__/mcp-tool-executor.service.spec.ts +++ b/packages/twenty-server/src/engine/api/mcp/services/__tests__/mcp-tool-executor.service.spec.ts @@ -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, + }, + }); + }); }); }); diff --git a/packages/twenty-server/src/engine/api/mcp/services/mcp-protocol.service.ts b/packages/twenty-server/src/engine/api/mcp/services/mcp-protocol.service.ts index 2eef4138d0..c83cbe94c9 100644 --- a/packages/twenty-server/src/engine/api/mcp/services/mcp-protocol.service.ts +++ b/packages/twenty-server/src/engine/api/mcp/services/mcp-protocol.service.ts @@ -178,6 +178,7 @@ export class McpProtocolService { userWorkspaceId?: string; apiKey: FlatApiKey | undefined; }, + sseWriter?: (data: Record) => void, ): Promise | null> { try { // JSON-RPC notifications have no id and expect no response @@ -244,6 +245,7 @@ export class McpProtocolService { id, toolSet, params, + sseWriter, ); } diff --git a/packages/twenty-server/src/engine/api/mcp/services/mcp-tool-executor.service.ts b/packages/twenty-server/src/engine/api/mcp/services/mcp-tool-executor.service.ts index bc14645ce3..59e3ce52eb 100644 --- a/packages/twenty-server/src/engine/api/mcp/services/mcp-tool-executor.service.ts +++ b/packages/twenty-server/src/engine/api/mcp/services/mcp-tool-executor.service.ts @@ -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, + sseWriter?: (data: Record) => 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', diff --git a/packages/twenty-server/src/engine/api/mcp/utils/__tests__/write-sse-event.util.spec.ts b/packages/twenty-server/src/engine/api/mcp/utils/__tests__/write-sse-event.util.spec.ts new file mode 100644 index 0000000000..03c7eb0df6 --- /dev/null +++ b/packages/twenty-server/src/engine/api/mcp/utils/__tests__/write-sse-event.util.spec.ts @@ -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); + }); +}); diff --git a/packages/twenty-server/src/engine/api/mcp/utils/write-sse-event.util.ts b/packages/twenty-server/src/engine/api/mcp/utils/write-sse-event.util.ts new file mode 100644 index 0000000000..899551aa71 --- /dev/null +++ b/packages/twenty-server/src/engine/api/mcp/utils/write-sse-event.util.ts @@ -0,0 +1,14 @@ +import { type Response } from 'express'; + +export const writeSseEvent = ( + response: Response, + data: Record, +): 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`); +}; diff --git a/packages/twenty-server/test/integration/ai/suites/mcp.controller.integration-spec.ts b/packages/twenty-server/test/integration/ai/suites/mcp.controller.integration-spec.ts index 6e9bd884c4..3c4d94eb7c 100644 --- a/packages/twenty-server/test/integration/ai/suites/mcp.controller.integration-spec.ts +++ b/packages/twenty-server/test/integration/ai/suites/mcp.controller.integration-spec.ts @@ -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> => { + 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); + }); + }); });