modelcontextprotocol/typescript-sdk
Watch
1
0
Fork
You've already forked typescript-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Merge branch 'ihrpr/streamable-http-client' into ihrpr/examples

Browse source
This commit is contained in:
ihrpr 2025-04-08 21:40:20 +01:00
commit 6c5efefd4a
No known key found for this signature in database
2 changed files with 21 additions and 111 deletions
Download patch file Download diff file Expand all files Collapse all files

40
src/client/streamableHttp.test.ts
View file

@ -124,46 +124,6 @@ describe("StreamableHTTPClientTransport", () => {
expect(errorSpy).toHaveBeenCalled();
});
it("should handle session termination via DELETE request", async () => {
// First set the session ID by mocking initialization
(global.fetch as jest.Mock).mockResolvedValueOnce({
ok: true,
status: 200,
headers: new Headers({ "mcp-session-id": "session-to-terminate" }),
});
await transport.send({
jsonrpc: "2.0",
method: "initialize",
params: {
clientInfo: { name: "test-client", version: "1.0" },
protocolVersion: "2025-03-26"
},
id: "init-id"
} as JSONRPCMessage);
// Mock DELETE request for session termination
(global.fetch as jest.Mock).mockResolvedValueOnce({
ok: true,
status: 200,
headers: new Headers()
});
const closeSpy = jest.fn();
transport.onclose = closeSpy;
await transport.close();
// Check that DELETE request was sent
const calls = (global.fetch as jest.Mock).mock.calls;
const lastCall = calls[calls.length - 1];
expect(lastCall[1].method).toBe("DELETE");
// The headers may be a plain object in tests
expect(lastCall[1].headers["mcp-session-id"]).toBe("session-to-terminate");
expect(closeSpy).toHaveBeenCalled();
});
it("should handle non-streaming JSON response", async () => {
const message: JSONRPCMessage = {
jsonrpc: "2.0",

92
src/client/streamableHttp.ts
View file

@ -2,7 +2,7 @@ import { Transport } from "../shared/transport.js";
import { JSONRPCMessage, JSONRPCMessageSchema } from "../types.js";
import { auth, AuthResult, OAuthClientProvider, UnauthorizedError } from "./auth.js";
import { type ErrorEvent } from "eventsource";
import { EventSourceParserStream } from 'eventsource-parser/stream';
export class StreamableHTTPError extends Error {
constructor(
public readonly code: number | undefined,
@ -45,7 +45,7 @@ export type StreamableHTTPClientTransportOptions = {
* for receiving messages.
*/
export class StreamableHTTPClientTransport implements Transport {
private _activeStreams: Map<string, ReadableStreamDefaultReader<Uint8Array>> = new Map();
private _activeStreams: Map<string, ReadableStreamDefaultReader<any>> = new Map();
private _abortController?: AbortController;
private _url: URL;
private _requestInit?: RequestInit;
@ -186,30 +186,6 @@ export class StreamableHTTPClientTransport implements Transport {
// Abort any pending requests
this._abortController?.abort();
// If we have a session ID, send a DELETE request to explicitly terminate the session
if (this._sessionId) {
try {
const commonHeaders = await this._commonHeaders();
const response = await fetch(this._url, {
method: "DELETE",
headers: commonHeaders,
signal: this._abortController?.signal,
});
if (!response.ok) {
// Server might respond with 405 if it doesn't support explicit session termination
// We don't throw an error in that case
if (response.status !== 405) {
const text = await response.text().catch(() => null);
throw new Error(`Error terminating session (HTTP ${response.status}): ${text}`);
}
}
} catch (error) {
// We still want to invoke onclose even if the session termination fails
this.onerror?.(error as Error);
}
}
this.onclose?.();
}
@ -300,62 +276,36 @@ export class StreamableHTTPClientTransport implements Transport {
return;
}
// Set up stream handling for server-sent events
const reader = stream.getReader();
// Create a pipeline: binary stream -> text decoder -> SSE parser
const eventStream = stream
.pipeThrough(new TextDecoderStream())
.pipeThrough(new EventSourceParserStream());
const reader = eventStream.getReader();
this._activeStreams.set(streamId, reader);
const decoder = new TextDecoder();
let buffer = '';
const processStream = async () => {
try {
while (true) {
const { done, value } = await reader.read();
const { done, value: event } = await reader.read();
if (done) {
// Stream closed by server
this._activeStreams.delete(streamId);
break;
}
buffer += decoder.decode(value, { stream: true });
// Update last event ID if provided
if (event.id) {
this._lastEventId = event.id;
}
// Process SSE messages in the buffer
const events = buffer.split('\n\n');
buffer = events.pop() || '';
for (const event of events) {
const lines = event.split('\n');
let id: string | undefined;
let eventType: string | undefined;
let data: string | undefined;
// Parse SSE message according to the format
for (const line of lines) {
if (line.startsWith('id:')) {
id = line.slice(3).trim();
} else if (line.startsWith('event:')) {
eventType = line.slice(6).trim();
} else if (line.startsWith('data:')) {
data = line.slice(5).trim();
}
}
// Update last event ID if provided by server
// As per spec: the ID MUST be globally unique across all streams within that session
if (id) {
this._lastEventId = id;
}
// Handle message event
if (data) {
// Default event type is 'message' per SSE spec if not specified
if (!eventType || eventType === 'message') {
try {
const message = JSONRPCMessageSchema.parse(JSON.parse(data));
this.onmessage?.(message);
} catch (error) {
this.onerror?.(error as Error);
}
}
// Handle message events (default event type is undefined per docs)
// or explicit 'message' event type
if (!event.event || event.event === 'message') {
try {
const message = JSONRPCMessageSchema.parse(JSON.parse(event.data));
this.onmessage?.(message);
} catch (error) {
this.onerror?.(error as Error);
}
}
}