Skip to content

Commit fe85673

Browse files
erwinmsmitherwinmsmith
committed
fix(mcp): honor connection abort signals
1 parent ac80a8f commit fe85673

2 files changed

Lines changed: 173 additions & 29 deletions

File tree

‎packages/mcp/src/connection-manager.ts‎

Lines changed: 41 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -1157,39 +1157,46 @@ class SDKMCPConnectionSession implements MCPConnectionSession {
11571157
private readonly options: SDKMCPConnectionSessionFactoryOptions
11581158
) {}
11591159

1160-
async connect(): Promise<{
1160+
async connect(signal?: AbortSignal): Promise<{
11611161
negotiatedProtocolVersion?: string;
11621162
serverInfo?: Record<string, unknown>;
11631163
serverCapabilities?: Record<string, unknown>;
11641164
}> {
1165+
signal?.throwIfAborted();
11651166
this.transport = await this.createTransport();
1166-
const versionedTransport = this.transport as typeof this.transport & {
1167-
setProtocolVersion?: (version: string) => void;
1168-
};
1169-
const setProtocolVersion = versionedTransport.setProtocolVersion?.bind(versionedTransport);
1170-
versionedTransport.setProtocolVersion = (version: string) => {
1171-
this.negotiatedProtocolVersion = version;
1172-
setProtocolVersion?.(version);
1173-
};
1174-
this.client = new Client(this.options.clientInfo ?? { name: 'hypha', version: '1.0.0' }, {
1175-
capabilities: {},
1176-
enforceStrictCapabilities: true,
1177-
listChanged: {
1178-
tools: { onChanged: () => this.onListChanged?.() },
1179-
resources: { onChanged: () => this.onListChanged?.() },
1180-
prompts: { onChanged: () => this.onListChanged?.() },
1181-
},
1182-
});
1183-
this.client.onclose = () => this.onClose?.();
1184-
this.client.onerror = (error) => this.onClose?.(error);
1185-
await this.client.connect(this.transport);
1186-
return {
1187-
negotiatedProtocolVersion: this.negotiatedProtocolVersion,
1188-
serverInfo: this.client.getServerVersion() as Record<string, unknown> | undefined,
1189-
serverCapabilities: this.client.getServerCapabilities() as
1190-
| Record<string, unknown>
1191-
| undefined,
1192-
};
1167+
try {
1168+
signal?.throwIfAborted();
1169+
const versionedTransport = this.transport as typeof this.transport & {
1170+
setProtocolVersion?: (version: string) => void;
1171+
};
1172+
const setProtocolVersion = versionedTransport.setProtocolVersion?.bind(versionedTransport);
1173+
versionedTransport.setProtocolVersion = (version: string) => {
1174+
this.negotiatedProtocolVersion = version;
1175+
setProtocolVersion?.(version);
1176+
};
1177+
this.client = new Client(this.options.clientInfo ?? { name: 'hypha', version: '1.0.0' }, {
1178+
capabilities: {},
1179+
enforceStrictCapabilities: true,
1180+
listChanged: {
1181+
tools: { onChanged: () => this.onListChanged?.() },
1182+
resources: { onChanged: () => this.onListChanged?.() },
1183+
prompts: { onChanged: () => this.onListChanged?.() },
1184+
},
1185+
});
1186+
this.client.onclose = () => this.onClose?.();
1187+
this.client.onerror = (error) => this.onClose?.(error);
1188+
await this.client.connect(this.transport, signal ? { signal } : undefined);
1189+
return {
1190+
negotiatedProtocolVersion: this.negotiatedProtocolVersion,
1191+
serverInfo: this.client.getServerVersion() as Record<string, unknown> | undefined,
1192+
serverCapabilities: this.client.getServerCapabilities() as
1193+
| Record<string, unknown>
1194+
| undefined,
1195+
};
1196+
} catch (error) {
1197+
await this.close().catch(() => undefined);
1198+
throw error;
1199+
}
11931200
}
11941201

11951202
async listCapabilities(signal?: AbortSignal): Promise<MCPCapabilityDescriptor[]> {
@@ -1369,10 +1376,15 @@ class SDKMCPConnectionSession implements MCPConnectionSession {
13691376

13701377
async close(): Promise<void> {
13711378
const client = this.client;
1379+
const transport = this.transport;
13721380
this.client = undefined;
1373-
if (client) await client.close();
13741381
this.transport = undefined;
13751382
this.negotiatedProtocolVersion = undefined;
1383+
if (client) {
1384+
await client.close();
1385+
return;
1386+
}
1387+
if (transport) await transport.close();
13761388
}
13771389

13781390
private async createTransport(): Promise<StdioClientTransport | StreamableHTTPClientTransport> {

‎packages/mcp/src/mcp.test.ts‎

Lines changed: 132 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
import { describe, expect, it } from 'vitest';
22
import express from 'express';
3+
import { once } from 'node:events';
4+
import { createServer } from 'node:http';
35
import { InMemoryEventStore, InMemoryTelemetryRecorder } from '@codesoul-co/hypha-core';
46
import {
57
GovernedToolRunner,
@@ -40,6 +42,85 @@ import {
4042
type MCPServerProfile,
4143
} from './index';
4244

45+
async function createInitializationFixture(responseDelayMs = 0): Promise<{
46+
endpoint: string;
47+
initializeRequests: () => number;
48+
waitForInitialize: () => Promise<void>;
49+
close: () => Promise<void>;
50+
}> {
51+
let initializeRequests = 0;
52+
let notifyInitialize: (() => void) | undefined;
53+
const initializeReceived = new Promise<void>((resolve) => {
54+
notifyInitialize = resolve;
55+
});
56+
const httpServer = createServer(async (request, response) => {
57+
if (request.method !== 'POST') {
58+
response.writeHead(405).end();
59+
return;
60+
}
61+
const chunks: Buffer[] = [];
62+
for await (const chunk of request) {
63+
chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk));
64+
}
65+
const message = JSON.parse(Buffer.concat(chunks).toString()) as {
66+
id?: string | number;
67+
method?: string;
68+
};
69+
if (message.method !== 'initialize') {
70+
response.writeHead(202).end();
71+
return;
72+
}
73+
initializeRequests += 1;
74+
notifyInitialize?.();
75+
if (responseDelayMs > 0) {
76+
await new Promise((resolve) => setTimeout(resolve, responseDelayMs));
77+
}
78+
if (response.destroyed) return;
79+
response.writeHead(200, { 'content-type': 'application/json' });
80+
response.end(
81+
JSON.stringify({
82+
jsonrpc: '2.0',
83+
id: message.id,
84+
result: {
85+
protocolVersion: '2025-11-25',
86+
capabilities: {},
87+
serverInfo: { name: 'abort-fixture', version: '1.0.0' },
88+
},
89+
})
90+
);
91+
});
92+
httpServer.listen(0, '127.0.0.1');
93+
await once(httpServer, 'listening');
94+
const address = httpServer.address();
95+
if (!address || typeof address === 'string') {
96+
throw new Error('Initialization fixture server has no port.');
97+
}
98+
return {
99+
endpoint: `http://127.0.0.1:${address.port}/mcp`,
100+
initializeRequests: () => initializeRequests,
101+
waitForInitialize: () => initializeReceived,
102+
close: async () => {
103+
httpServer.closeAllConnections();
104+
await new Promise<void>((resolve, reject) =>
105+
httpServer.close((error) => (error ? reject(error) : resolve()))
106+
);
107+
},
108+
};
109+
}
110+
111+
function createSDKHTTPConnectionSession(endpoint: string): MCPConnectionSession {
112+
return new SDKMCPConnectionSessionFactory().create({
113+
id: 'abort-fixture',
114+
mode: 'fixture',
115+
transport: { type: 'streamable_http', endpoint },
116+
egressPolicy: {
117+
allowedHosts: ['127.0.0.1'],
118+
denyPrivateNetworks: false,
119+
requireTls: false,
120+
},
121+
});
122+
}
123+
43124
describe('@codesoul-co/hypha-mcp normalization', () => {
44125
it('unwraps structured Tool output without misclassifying domain objects', () => {
45126
expect(
@@ -1331,6 +1412,57 @@ describe('@codesoul-co/hypha-mcp normalization', () => {
13311412
await expect(manager.get('stdio-fixture')).resolves.toMatchObject({ state: 'closed' });
13321413
});
13331414

1415+
it('rejects an already-aborted signal before SDK initialization starts', async () => {
1416+
const fixture = await createInitializationFixture();
1417+
const session = createSDKHTTPConnectionSession(fixture.endpoint);
1418+
const controller = new AbortController();
1419+
controller.abort(new Error('cancelled before connect'));
1420+
1421+
try {
1422+
await expect(session.connect(controller.signal)).rejects.toThrow('cancelled before connect');
1423+
expect(fixture.initializeRequests()).toBe(0);
1424+
} finally {
1425+
await session.close().catch(() => undefined);
1426+
await fixture.close();
1427+
}
1428+
});
1429+
1430+
it('aborts and cleans up SDK initialization while the response is pending', async () => {
1431+
const fixture = await createInitializationFixture(150);
1432+
const session = createSDKHTTPConnectionSession(fixture.endpoint);
1433+
const controller = new AbortController();
1434+
const connecting = session.connect(controller.signal);
1435+
1436+
try {
1437+
await fixture.waitForInitialize();
1438+
controller.abort(new Error('cancelled during connect'));
1439+
await expect(connecting).rejects.toThrow('cancelled during connect');
1440+
expect(fixture.initializeRequests()).toBe(1);
1441+
await expect(session.connect()).resolves.toMatchObject({
1442+
negotiatedProtocolVersion: '2025-11-25',
1443+
});
1444+
} finally {
1445+
await session.close().catch(() => undefined);
1446+
await fixture.close();
1447+
}
1448+
});
1449+
1450+
it('preserves SDK initialization when no cancellation signal is provided', async () => {
1451+
const fixture = await createInitializationFixture();
1452+
const session = createSDKHTTPConnectionSession(fixture.endpoint);
1453+
1454+
try {
1455+
await expect(session.connect()).resolves.toMatchObject({
1456+
negotiatedProtocolVersion: '2025-11-25',
1457+
serverInfo: { name: 'abort-fixture', version: '1.0.0' },
1458+
});
1459+
expect(fixture.initializeRequests()).toBe(1);
1460+
} finally {
1461+
await session.close().catch(() => undefined);
1462+
await fixture.close();
1463+
}
1464+
});
1465+
13341466
it('connects through the real MCP Streamable HTTP transport', async () => {
13351467
const fixtureErrors: string[] = [];
13361468
const createProtocolServer = () => {

0 commit comments

Comments
 (0)