diff --git a/CHANGELOG.md b/CHANGELOG.md index b689b2f..0cf1342 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ ### CLI - Correct the documented call timeout to 60 seconds and distinguish its `MCPORTER_CALL_TIMEOUT` and `--timeout` overrides from the 30-second list timeout. (PR #230, thanks @KrasimirKralev) +- Isolate long-lived SSE receive streams from ordinary HTTP requests so byte-idle servers cannot stall tool listing and calls. (PR #234, thanks @umutkeltek) ### OAuth diff --git a/docs/config.md b/docs/config.md index 0f14d7e..6a5c534 100644 --- a/docs/config.md +++ b/docs/config.md @@ -168,7 +168,9 @@ Use `--scope home|project` with `mcporter config add` to pick the write target e ## HTTP Compatibility -HTTP MCP servers normally use Node's built-in `fetch` through the upstream MCP SDK. If a provider rejects that stack but accepts plain Node `https.request` traffic, set `httpFetch: "node-http1"` on the server entry to force an HTTP/1.1 fetch implementation for Streamable HTTP and SSE POST requests: +HTTP MCP servers normally use Node's built-in `fetch` for request/response traffic. MCPorter's optional long-lived SSE receive channel uses a separate HTTP/1.1 connection so an idle stream cannot occupy the request pool and stall later tool calls. Set `httpFetch: "default"` to opt out of that isolation and use the built-in fetch for every request. + +If a provider rejects the built-in stack entirely but accepts plain Node `https.request` traffic, set `httpFetch: "node-http1"` on the server entry to force an HTTP/1.1 fetch implementation for all Streamable HTTP and SSE requests: ```jsonc { @@ -182,7 +184,7 @@ HTTP MCP servers normally use Node's built-in `fetch` through the upstream MCP S } ``` -The Sunsama endpoint is auto-detected and uses this compatibility path by default. +The Sunsama endpoint is auto-detected and uses the all-request compatibility path by default. ## JSON Schema for IDE Support diff --git a/src/runtime/node-http-fetch.ts b/src/runtime/node-http-fetch.ts index e9e3b25..cc4a1c1 100644 --- a/src/runtime/node-http-fetch.ts +++ b/src/runtime/node-http-fetch.ts @@ -12,6 +12,24 @@ export const nodeHttp1Fetch: FetchLike = async (input, init = {}) => { return nodeHttp1FetchWithRedirects(input, init, 0); }; +/** + * Keep the optional, long-lived SSE receive channel off the fetch implementation + * used for ordinary MCP requests. Some fetch stacks can otherwise let an idle GET + * monopolize the origin pool and indefinitely queue later POST requests. + */ +export function createSseIsolatedFetch(defaultFetch: FetchLike): FetchLike { + return (input, init = {}) => { + const method = (init.method ?? 'GET').toUpperCase(); + const acceptsSse = new Headers(init.headers).get('accept')?.toLowerCase().includes('text/event-stream') ?? false; + if (method === 'GET' && acceptsSse) { + return nodeHttp1Fetch(input, init); + } + return defaultFetch(input, init); + }; +} + +export const sseIsolatedFetch = createSseIsolatedFetch((input, init) => fetch(input, init)); + async function nodeHttp1FetchWithRedirects( input: string | URL, init: RequestInit, diff --git a/src/runtime/transport.ts b/src/runtime/transport.ts index 4befeba..a6c1b57 100644 --- a/src/runtime/transport.ts +++ b/src/runtime/transport.ts @@ -13,7 +13,7 @@ import { readCachedAccessToken } from '../oauth-persistence.js'; import { materializeHeaders } from '../runtime-header-utils.js'; import { isUnauthorizedError, maybeEnableOAuth } from '../runtime-oauth-support.js'; import { closeTransportAndWait } from '../runtime-process-utils.js'; -import { nodeHttp1Fetch } from './node-http-fetch.js'; +import { nodeHttp1Fetch, sseIsolatedFetch } from './node-http-fetch.js'; import { connectWithAuth, isOAuthFlowError, @@ -128,6 +128,8 @@ function createHttpTransportOptions( }; } +const NODE_HTTP1_FETCH_HOSTS: ReadonlySet = new Set(['api.sunsama.com']); + function resolveHttpFetchOverride(definition: ServerDefinition): typeof nodeHttp1Fetch | undefined { if (definition.command.kind !== 'http') { return undefined; @@ -138,10 +140,13 @@ function resolveHttpFetchOverride(definition: ServerDefinition): typeof nodeHttp if (definition.httpFetch === 'node-http1') { return nodeHttp1Fetch; } - if (definition.command.url.hostname.toLowerCase() === 'api.sunsama.com') { + if (NODE_HTTP1_FETCH_HOSTS.has(definition.command.url.hostname.toLowerCase())) { return nodeHttp1Fetch; } - return undefined; + if ('bun' in process.versions) { + return undefined; + } + return sseIsolatedFetch; } async function closeOAuthSession(oauthSession?: OAuthSession): Promise { diff --git a/tests/cli-idle-sse.integration.test.ts b/tests/cli-idle-sse.integration.test.ts new file mode 100644 index 0000000..6b742cc --- /dev/null +++ b/tests/cli-idle-sse.integration.test.ts @@ -0,0 +1,199 @@ +import { execFile } from 'node:child_process'; +import fs from 'node:fs/promises'; +import { createServer, type Server as HttpServer, type ServerResponse } from 'node:http'; +import type { AddressInfo } from 'node:net'; +import os from 'node:os'; +import path from 'node:path'; +import process from 'node:process'; +import { fileURLToPath, pathToFileURL } from 'node:url'; +import { afterAll, beforeAll, describe, expect, it } from 'vitest'; + +const CLI_ENTRY = fileURLToPath(new URL('../dist/cli.js', import.meta.url)); +const SINGLE_CONNECTION_FETCH = String.raw` +import http from 'node:http'; +import { Readable } from 'node:stream'; + +const agent = new http.Agent({ keepAlive: true, maxSockets: 1 }); + +globalThis.fetch = async (input, init = {}) => { + const url = input instanceof URL ? input : new URL(input); + if (url.protocol !== 'http:') { + throw new TypeError('single-connection test fetch only supports http URLs'); + } + const headers = Object.fromEntries(new Headers(init.headers)); + const body = init.body == null ? undefined : String(init.body); + if (body !== undefined && !Object.keys(headers).some((key) => key.toLowerCase() === 'content-length')) { + headers['content-length'] = String(Buffer.byteLength(body)); + } + + return new Promise((resolve, reject) => { + const request = http.request(url, { method: init.method ?? 'GET', headers, agent }, (response) => { + const responseHeaders = new Headers(); + for (const [key, value] of Object.entries(response.headers)) { + if (Array.isArray(value)) { + for (const item of value) responseHeaders.append(key, item); + } else if (value !== undefined) { + responseHeaders.set(key, String(value)); + } + } + resolve(new Response(Readable.toWeb(response), { + status: response.statusCode ?? 502, + statusText: response.statusMessage, + headers: responseHeaders, + })); + }); + const abort = () => request.destroy(new DOMException('The operation was aborted.', 'AbortError')); + init.signal?.addEventListener('abort', abort, { once: true }); + request.once('close', () => init.signal?.removeEventListener('abort', abort)); + request.once('error', reject); + request.end(body); + }); +}; +`; + +interface CliResult { + readonly stdout: string; + readonly stderr: string; +} + +function runCli(args: string[], configPath: string, preloadPath: string): Promise { + return new Promise((resolve, reject) => { + execFile( + process.execPath, + [CLI_ENTRY, '--config', configPath, ...args], + { + cwd: process.cwd(), + env: { + ...process.env, + MCPORTER_NO_FORCE_EXIT: '1', + NODE_OPTIONS: `--import=${pathToFileURL(preloadPath).href}`, + }, + maxBuffer: 1024 * 1024, + timeout: 10_000, + }, + (error, stdout, stderr) => { + if (error) { + reject(new Error(`${error.message}\nSTDOUT:\n${stdout}\nSTDERR:\n${stderr}`)); + return; + } + resolve({ stdout, stderr }); + } + ); + }); +} + +describe('idle standalone SSE CLI integration', () => { + let httpServer: HttpServer; + let idleResponse: ServerResponse | undefined; + let configPath: string; + let preloadPath: string; + let tempDir: string; + const requestOrder: string[] = []; + let toolsListSawOpenSse = false; + + beforeAll(async () => { + await fs.access(CLI_ENTRY); + + httpServer = createServer(async (request, response) => { + if (request.method === 'GET') { + requestOrder.push('GET standalone-sse'); + idleResponse = response; + response.writeHead(200, { + 'cache-control': 'no-cache', + connection: 'keep-alive', + 'content-type': 'text/event-stream', + }); + response.flushHeaders(); + return; + } + + let body = ''; + request.setEncoding('utf8'); + for await (const chunk of request) { + body += chunk; + } + const message = JSON.parse(body) as { id?: number; method: string }; + requestOrder.push(message.method); + + if (message.method === 'initialize') { + response.writeHead(200, { + 'content-type': 'application/json', + 'mcp-session-id': 'idle-sse-session', + }); + response.end( + JSON.stringify({ + jsonrpc: '2.0', + id: message.id, + result: { + capabilities: { tools: {} }, + protocolVersion: '2025-11-25', + serverInfo: { name: 'idle-sse-test', version: '1.0.0' }, + }, + }) + ); + return; + } + + if (message.method === 'notifications/initialized') { + response.writeHead(202).end(); + return; + } + + if (message.method === 'tools/list') { + toolsListSawOpenSse = Boolean(idleResponse && !idleResponse.writableEnded && !idleResponse.destroyed); + response.writeHead(200, { 'content-type': 'application/json' }); + response.end( + JSON.stringify({ + jsonrpc: '2.0', + id: message.id, + result: { + tools: [ + { + name: 'ping', + description: 'Return pong', + inputSchema: { type: 'object', properties: {} }, + }, + ], + }, + }) + ); + return; + } + + response.writeHead(404).end(); + }); + + await new Promise((resolve, reject) => { + httpServer.once('error', reject); + httpServer.listen(0, '127.0.0.1', resolve); + }); + const address = httpServer.address() as AddressInfo; + const baseUrl = `http://127.0.0.1:${address.port}/mcp`; + + tempDir = await fs.mkdtemp(path.join(os.tmpdir(), 'mcporter-idle-sse-')); + configPath = path.join(tempDir, 'config.json'); + preloadPath = path.join(tempDir, 'single-connection-fetch.mjs'); + await fs.writeFile(configPath, JSON.stringify({ imports: [], mcpServers: { idle: { baseUrl } } }, null, 2), 'utf8'); + await fs.writeFile(preloadPath, SINGLE_CONNECTION_FETCH, 'utf8'); + }); + + afterAll(async () => { + idleResponse?.end(); + await new Promise((resolve) => httpServer.close(() => resolve())); + await fs.rm(tempDir, { recursive: true, force: true }); + }); + + it('lists tools while a standalone SSE stream is open and byte-idle', async () => { + const result = await runCli(['list', 'idle', '--json', '--timeout', '2000'], configPath, preloadPath); + + expect(result.stderr).toBe(''); + expect(JSON.parse(result.stdout)).toMatchObject({ + mode: 'server', + name: 'idle', + status: 'ok', + tools: [expect.objectContaining({ name: 'ping' })], + }); + expect(requestOrder).toEqual(['initialize', 'notifications/initialized', 'GET standalone-sse', 'tools/list']); + expect(toolsListSawOpenSse).toBe(true); + }); +}); diff --git a/tests/generate-cli.test.ts b/tests/generate-cli.test.ts index 2b4d096..3ad9b4a 100644 --- a/tests/generate-cli.test.ts +++ b/tests/generate-cli.test.ts @@ -218,7 +218,8 @@ describeGenerateCli('generateCli', () => { } = await generateCli({ serverRef: inline, runtime: 'bun', - timeoutMs: 5_000, + // Bun's first compiled invocation can spend several seconds initializing on loaded CI hosts. + timeoutMs: 10_000, minify: true, compile: expectedBinaryPath, }); diff --git a/tests/node-http-fetch.test.ts b/tests/node-http-fetch.test.ts index a606be2..6475ca8 100644 --- a/tests/node-http-fetch.test.ts +++ b/tests/node-http-fetch.test.ts @@ -1,7 +1,7 @@ import { createServer } from 'node:http'; import type { IncomingMessage, ServerResponse } from 'node:http'; -import { afterEach, describe, expect, it } from 'vitest'; -import { nodeHttp1Fetch } from '../src/runtime/node-http-fetch.js'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { createSseIsolatedFetch, nodeHttp1Fetch } from '../src/runtime/node-http-fetch.js'; let cleanup: (() => Promise) | undefined; @@ -11,6 +11,28 @@ afterEach(async () => { }); describe('nodeHttp1Fetch', () => { + it('isolates only event-stream GET requests from the default fetch', async () => { + const { baseUrl, close } = await serve((_request, response) => { + response.writeHead(200, { 'content-type': 'text/plain' }); + response.end('isolated'); + }); + cleanup = close; + const defaultFetch = vi.fn(async () => new Response('default')); + const isolatedFetch = createSseIsolatedFetch(defaultFetch); + + const postResponse = await isolatedFetch(baseUrl, { + method: 'POST', + headers: { accept: 'application/json, text/event-stream' }, + }); + const jsonGetResponse = await isolatedFetch(baseUrl, { headers: { accept: 'application/json' } }); + const sseGetResponse = await isolatedFetch(baseUrl, { headers: { accept: 'text/event-stream' } }); + + expect(await postResponse.text()).toBe('default'); + expect(await jsonGetResponse.text()).toBe('default'); + expect(await sseGetResponse.text()).toBe('isolated'); + expect(defaultFetch).toHaveBeenCalledTimes(2); + }); + it('follows redirects by default', async () => { const { baseUrl, close } = await serve((request, response) => { if (request.url === '/start') { diff --git a/tests/runtime-transport.test.ts b/tests/runtime-transport.test.ts index b2b2f73..c765841 100644 --- a/tests/runtime-transport.test.ts +++ b/tests/runtime-transport.test.ts @@ -325,6 +325,18 @@ describe('createClientContext (HTTP)', () => { await createClientContext(definition, logger, clientInfo, { maxOAuthAttempts: 0 }); }); + it('isolates the standalone SSE channel from the default fetch pool', async () => { + const definition = stubHttpDefinition('https://example.com/mcp'); + + vi.spyOn(Client.prototype, 'connect').mockImplementationOnce(async (transport) => { + expect(transport).toBeInstanceOf(StreamableHTTPClientTransport); + const fetchOverride = (transport as { _fetch?: unknown })._fetch; + expect(fetchOverride).toEqual(expect.any(Function)); + }); + + await createClientContext(definition, logger, clientInfo, { maxOAuthAttempts: 0 }); + }); + it('uses the HTTP/1.1 fetch compatibility path for Sunsama by default', async () => { const definition = stubHttpDefinition('https://api.sunsama.com/mcp'); @@ -352,6 +364,21 @@ describe('createClientContext (HTTP)', () => { await createClientContext(definition, logger, clientInfo, { maxOAuthAttempts: 0 }); }); + it('honors explicit default fetch mode for other HTTP servers', async () => { + const definition: ServerDefinition = { + ...stubHttpDefinition('https://example.com/mcp'), + httpFetch: 'default', + }; + + vi.spyOn(Client.prototype, 'connect').mockImplementationOnce(async (transport) => { + expect(transport).toBeInstanceOf(StreamableHTTPClientTransport); + const fetchOverride = (transport as { _fetch?: unknown })._fetch; + expect(fetchOverride).toBeUndefined(); + }); + + await createClientContext(definition, logger, clientInfo, { maxOAuthAttempts: 0 }); + }); + it('does not create OAuth sessions for OAuth HTTP servers when disableOAuth is true', async () => { const definition = stubOAuthHttpDefinition('https://example.com/secure');