Merge remote-tracking branch 'upstream/main' into request-auth-info
This commit is contained in:
Allen Zhou
commit
6f852a1117
6 changed files with 5074 additions and 102 deletions
File diff suppressed because it is too large
Load diff
|
|
@ -37,6 +37,8 @@ import {
|
|||
PromptArgument,
|
||||
GetPromptResult,
|
||||
ReadResourceResult,
|
||||
ServerRequest,
|
||||
ServerNotification,
|
||||
} from "../types.js";
|
||||
import { Completable, CompletableDef } from "./completable.js";
|
||||
import { UriTemplate, Variables } from "../shared/uriTemplate.js";
|
||||
|
|
@ -96,13 +98,17 @@ export class McpServer {
|
|||
);
|
||||
|
||||
this.server.registerCapabilities({
|
||||
tools: {},
|
||||
});
|
||||
tools: {
|
||||
listChanged: true
|
||||
}
|
||||
})
|
||||
|
||||
this.server.setRequestHandler(
|
||||
ListToolsRequestSchema,
|
||||
(): ListToolsResult => ({
|
||||
tools: Object.entries(this._registeredTools).map(
|
||||
tools: Object.entries(this._registeredTools).filter(
|
||||
([, tool]) => tool.enabled,
|
||||
).map(
|
||||
([name, tool]): Tool => {
|
||||
return {
|
||||
name,
|
||||
|
|
@ -129,6 +135,13 @@ export class McpServer {
|
|||
);
|
||||
}
|
||||
|
||||
if (!tool.enabled) {
|
||||
throw new McpError(
|
||||
ErrorCode.InvalidParams,
|
||||
`Tool ${request.params.name} disabled`,
|
||||
);
|
||||
}
|
||||
|
||||
if (tool.inputSchema) {
|
||||
const parseResult = await tool.inputSchema.safeParseAsync(
|
||||
request.params.arguments,
|
||||
|
|
@ -218,7 +231,14 @@ export class McpServer {
|
|||
if (!prompt) {
|
||||
throw new McpError(
|
||||
ErrorCode.InvalidParams,
|
||||
`Prompt ${request.params.ref.name} not found`,
|
||||
`Prompt ${ref.name} not found`,
|
||||
);
|
||||
}
|
||||
|
||||
if (!prompt.enabled) {
|
||||
throw new McpError(
|
||||
ErrorCode.InvalidParams,
|
||||
`Prompt ${ref.name} disabled`,
|
||||
);
|
||||
}
|
||||
|
||||
|
|
@ -285,13 +305,17 @@ export class McpServer {
|
|||
);
|
||||
|
||||
this.server.registerCapabilities({
|
||||
resources: {},
|
||||
});
|
||||
resources: {
|
||||
listChanged: true
|
||||
}
|
||||
})
|
||||
|
||||
this.server.setRequestHandler(
|
||||
ListResourcesRequestSchema,
|
||||
async (request, extra) => {
|
||||
const resources = Object.entries(this._registeredResources).map(
|
||||
const resources = Object.entries(this._registeredResources).filter(
|
||||
([_, resource]) => resource.enabled,
|
||||
).map(
|
||||
([uri, resource]) => ({
|
||||
uri,
|
||||
name: resource.name,
|
||||
|
|
@ -343,6 +367,12 @@ export class McpServer {
|
|||
// First check for exact resource match
|
||||
const resource = this._registeredResources[uri.toString()];
|
||||
if (resource) {
|
||||
if (!resource.enabled) {
|
||||
throw new McpError(
|
||||
ErrorCode.InvalidParams,
|
||||
`Resource ${uri} disabled`,
|
||||
);
|
||||
}
|
||||
return resource.readCallback(uri, extra);
|
||||
}
|
||||
|
||||
|
|
@ -385,13 +415,17 @@ export class McpServer {
|
|||
);
|
||||
|
||||
this.server.registerCapabilities({
|
||||
prompts: {},
|
||||
});
|
||||
prompts: {
|
||||
listChanged: true
|
||||
}
|
||||
})
|
||||
|
||||
this.server.setRequestHandler(
|
||||
ListPromptsRequestSchema,
|
||||
(): ListPromptsResult => ({
|
||||
prompts: Object.entries(this._registeredPrompts).map(
|
||||
prompts: Object.entries(this._registeredPrompts).filter(
|
||||
([, prompt]) => prompt.enabled,
|
||||
).map(
|
||||
([name, prompt]): Prompt => {
|
||||
return {
|
||||
name,
|
||||
|
|
@ -416,6 +450,13 @@ export class McpServer {
|
|||
);
|
||||
}
|
||||
|
||||
if (!prompt.enabled) {
|
||||
throw new McpError(
|
||||
ErrorCode.InvalidParams,
|
||||
`Prompt ${request.params.name} disabled`,
|
||||
);
|
||||
}
|
||||
|
||||
if (prompt.argsSchema) {
|
||||
const parseResult = await prompt.argsSchema.safeParseAsync(
|
||||
request.params.arguments,
|
||||
|
|
@ -445,7 +486,7 @@ export class McpServer {
|
|||
/**
|
||||
* Registers a resource `name` at a fixed URI, which will use the given callback to respond to read requests.
|
||||
*/
|
||||
resource(name: string, uri: string, readCallback: ReadResourceCallback): void;
|
||||
resource(name: string, uri: string, readCallback: ReadResourceCallback): RegisteredResource;
|
||||
|
||||
/**
|
||||
* Registers a resource `name` at a fixed URI with metadata, which will use the given callback to respond to read requests.
|
||||
|
|
@ -455,7 +496,7 @@ export class McpServer {
|
|||
uri: string,
|
||||
metadata: ResourceMetadata,
|
||||
readCallback: ReadResourceCallback,
|
||||
): void;
|
||||
): RegisteredResource;
|
||||
|
||||
/**
|
||||
* Registers a resource `name` with a template pattern, which will use the given callback to respond to read requests.
|
||||
|
|
@ -464,7 +505,7 @@ export class McpServer {
|
|||
name: string,
|
||||
template: ResourceTemplate,
|
||||
readCallback: ReadResourceTemplateCallback,
|
||||
): void;
|
||||
): RegisteredResourceTemplate;
|
||||
|
||||
/**
|
||||
* Registers a resource `name` with a template pattern and metadata, which will use the given callback to respond to read requests.
|
||||
|
|
@ -474,13 +515,13 @@ export class McpServer {
|
|||
template: ResourceTemplate,
|
||||
metadata: ResourceMetadata,
|
||||
readCallback: ReadResourceTemplateCallback,
|
||||
): void;
|
||||
): RegisteredResourceTemplate;
|
||||
|
||||
resource(
|
||||
name: string,
|
||||
uriOrTemplate: string | ResourceTemplate,
|
||||
...rest: unknown[]
|
||||
): void {
|
||||
): RegisteredResource | RegisteredResourceTemplate {
|
||||
let metadata: ResourceMetadata | undefined;
|
||||
if (typeof rest[0] === "object") {
|
||||
metadata = rest.shift() as ResourceMetadata;
|
||||
|
|
@ -495,35 +536,73 @@ export class McpServer {
|
|||
throw new Error(`Resource ${uriOrTemplate} is already registered`);
|
||||
}
|
||||
|
||||
this._registeredResources[uriOrTemplate] = {
|
||||
const registeredResource: RegisteredResource = {
|
||||
name,
|
||||
metadata,
|
||||
readCallback: readCallback as ReadResourceCallback,
|
||||
enabled: true,
|
||||
disable: () => registeredResource.update({ enabled: false }),
|
||||
enable: () => registeredResource.update({ enabled: true }),
|
||||
remove: () => registeredResource.update({ uri: null }),
|
||||
update: (updates) => {
|
||||
if (typeof updates.uri !== "undefined" && updates.uri !== uriOrTemplate) {
|
||||
delete this._registeredResources[uriOrTemplate]
|
||||
if (updates.uri) this._registeredResources[updates.uri] = registeredResource
|
||||
}
|
||||
if (typeof updates.name !== "undefined") registeredResource.name = updates.name
|
||||
if (typeof updates.metadata !== "undefined") registeredResource.metadata = updates.metadata
|
||||
if (typeof updates.callback !== "undefined") registeredResource.readCallback = updates.callback
|
||||
if (typeof updates.enabled !== "undefined") registeredResource.enabled = updates.enabled
|
||||
this.sendResourceListChanged()
|
||||
},
|
||||
};
|
||||
this._registeredResources[uriOrTemplate] = registeredResource;
|
||||
|
||||
this.setResourceRequestHandlers();
|
||||
this.sendResourceListChanged();
|
||||
return registeredResource;
|
||||
} else {
|
||||
if (this._registeredResourceTemplates[name]) {
|
||||
throw new Error(`Resource template ${name} is already registered`);
|
||||
}
|
||||
|
||||
this._registeredResourceTemplates[name] = {
|
||||
const registeredResourceTemplate: RegisteredResourceTemplate = {
|
||||
resourceTemplate: uriOrTemplate,
|
||||
metadata,
|
||||
readCallback: readCallback as ReadResourceTemplateCallback,
|
||||
enabled: true,
|
||||
disable: () => registeredResourceTemplate.update({ enabled: false }),
|
||||
enable: () => registeredResourceTemplate.update({ enabled: true }),
|
||||
remove: () => registeredResourceTemplate.update({ name: null }),
|
||||
update: (updates) => {
|
||||
if (typeof updates.name !== "undefined" && updates.name !== name) {
|
||||
delete this._registeredResourceTemplates[name]
|
||||
if (updates.name) this._registeredResourceTemplates[updates.name] = registeredResourceTemplate
|
||||
}
|
||||
if (typeof updates.template !== "undefined") registeredResourceTemplate.resourceTemplate = updates.template
|
||||
if (typeof updates.metadata !== "undefined") registeredResourceTemplate.metadata = updates.metadata
|
||||
if (typeof updates.callback !== "undefined") registeredResourceTemplate.readCallback = updates.callback
|
||||
if (typeof updates.enabled !== "undefined") registeredResourceTemplate.enabled = updates.enabled
|
||||
this.sendResourceListChanged()
|
||||
},
|
||||
};
|
||||
}
|
||||
this._registeredResourceTemplates[name] = registeredResourceTemplate;
|
||||
|
||||
this.setResourceRequestHandlers();
|
||||
this.setResourceRequestHandlers();
|
||||
this.sendResourceListChanged();
|
||||
return registeredResourceTemplate;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Registers a zero-argument tool `name`, which will run the given function when the client calls it.
|
||||
*/
|
||||
tool(name: string, cb: ToolCallback): void;
|
||||
tool(name: string, cb: ToolCallback): RegisteredTool;
|
||||
|
||||
/**
|
||||
* Registers a zero-argument tool `name` (with a description) which will run the given function when the client calls it.
|
||||
*/
|
||||
tool(name: string, description: string, cb: ToolCallback): void;
|
||||
tool(name: string, description: string, cb: ToolCallback): RegisteredTool;
|
||||
|
||||
/**
|
||||
* Registers a tool `name` accepting the given arguments, which must be an object containing named properties associated with Zod schemas. When the client calls it, the function will be run with the parsed and validated arguments.
|
||||
|
|
@ -532,7 +611,7 @@ export class McpServer {
|
|||
name: string,
|
||||
paramsSchema: Args,
|
||||
cb: ToolCallback<Args>,
|
||||
): void;
|
||||
): RegisteredTool;
|
||||
|
||||
/**
|
||||
* Registers a tool `name` (with a description) accepting the given arguments, which must be an object containing named properties associated with Zod schemas. When the client calls it, the function will be run with the parsed and validated arguments.
|
||||
|
|
@ -542,9 +621,9 @@ export class McpServer {
|
|||
description: string,
|
||||
paramsSchema: Args,
|
||||
cb: ToolCallback<Args>,
|
||||
): void;
|
||||
): RegisteredTool;
|
||||
|
||||
tool(name: string, ...rest: unknown[]): void {
|
||||
tool(name: string, ...rest: unknown[]): RegisteredTool {
|
||||
if (this._registeredTools[name]) {
|
||||
throw new Error(`Tool ${name} is already registered`);
|
||||
}
|
||||
|
|
@ -560,25 +639,44 @@ export class McpServer {
|
|||
}
|
||||
|
||||
const cb = rest[0] as ToolCallback<ZodRawShape | undefined>;
|
||||
this._registeredTools[name] = {
|
||||
const registeredTool: RegisteredTool = {
|
||||
description,
|
||||
inputSchema:
|
||||
paramsSchema === undefined ? undefined : z.object(paramsSchema),
|
||||
callback: cb,
|
||||
enabled: true,
|
||||
disable: () => registeredTool.update({ enabled: false }),
|
||||
enable: () => registeredTool.update({ enabled: true }),
|
||||
remove: () => registeredTool.update({ name: null }),
|
||||
update: (updates) => {
|
||||
if (typeof updates.name !== "undefined" && updates.name !== name) {
|
||||
delete this._registeredTools[name]
|
||||
if (updates.name) this._registeredTools[updates.name] = registeredTool
|
||||
}
|
||||
if (typeof updates.description !== "undefined") registeredTool.description = updates.description
|
||||
if (typeof updates.paramsSchema !== "undefined") registeredTool.inputSchema = z.object(updates.paramsSchema)
|
||||
if (typeof updates.callback !== "undefined") registeredTool.callback = updates.callback
|
||||
if (typeof updates.enabled !== "undefined") registeredTool.enabled = updates.enabled
|
||||
this.sendToolListChanged()
|
||||
},
|
||||
};
|
||||
this._registeredTools[name] = registeredTool;
|
||||
|
||||
this.setToolRequestHandlers();
|
||||
this.sendToolListChanged()
|
||||
|
||||
return registeredTool
|
||||
}
|
||||
|
||||
/**
|
||||
* Registers a zero-argument prompt `name`, which will run the given function when the client calls it.
|
||||
*/
|
||||
prompt(name: string, cb: PromptCallback): void;
|
||||
prompt(name: string, cb: PromptCallback): RegisteredPrompt;
|
||||
|
||||
/**
|
||||
* Registers a zero-argument prompt `name` (with a description) which will run the given function when the client calls it.
|
||||
*/
|
||||
prompt(name: string, description: string, cb: PromptCallback): void;
|
||||
prompt(name: string, description: string, cb: PromptCallback): RegisteredPrompt;
|
||||
|
||||
/**
|
||||
* Registers a prompt `name` accepting the given arguments, which must be an object containing named properties associated with Zod schemas. When the client calls it, the function will be run with the parsed and validated arguments.
|
||||
|
|
@ -587,7 +685,7 @@ export class McpServer {
|
|||
name: string,
|
||||
argsSchema: Args,
|
||||
cb: PromptCallback<Args>,
|
||||
): void;
|
||||
): RegisteredPrompt;
|
||||
|
||||
/**
|
||||
* Registers a prompt `name` (with a description) accepting the given arguments, which must be an object containing named properties associated with Zod schemas. When the client calls it, the function will be run with the parsed and validated arguments.
|
||||
|
|
@ -597,9 +695,9 @@ export class McpServer {
|
|||
description: string,
|
||||
argsSchema: Args,
|
||||
cb: PromptCallback<Args>,
|
||||
): void;
|
||||
): RegisteredPrompt;
|
||||
|
||||
prompt(name: string, ...rest: unknown[]): void {
|
||||
prompt(name: string, ...rest: unknown[]): RegisteredPrompt {
|
||||
if (this._registeredPrompts[name]) {
|
||||
throw new Error(`Prompt ${name} is already registered`);
|
||||
}
|
||||
|
|
@ -615,13 +713,67 @@ export class McpServer {
|
|||
}
|
||||
|
||||
const cb = rest[0] as PromptCallback<PromptArgsRawShape | undefined>;
|
||||
this._registeredPrompts[name] = {
|
||||
const registeredPrompt: RegisteredPrompt = {
|
||||
description,
|
||||
argsSchema: argsSchema === undefined ? undefined : z.object(argsSchema),
|
||||
callback: cb,
|
||||
enabled: true,
|
||||
disable: () => registeredPrompt.update({ enabled: false }),
|
||||
enable: () => registeredPrompt.update({ enabled: true }),
|
||||
remove: () => registeredPrompt.update({ name: null }),
|
||||
update: (updates) => {
|
||||
if (typeof updates.name !== "undefined" && updates.name !== name) {
|
||||
delete this._registeredPrompts[name]
|
||||
if (updates.name) this._registeredPrompts[updates.name] = registeredPrompt
|
||||
}
|
||||
if (typeof updates.description !== "undefined") registeredPrompt.description = updates.description
|
||||
if (typeof updates.argsSchema !== "undefined") registeredPrompt.argsSchema = z.object(updates.argsSchema)
|
||||
if (typeof updates.callback !== "undefined") registeredPrompt.callback = updates.callback
|
||||
if (typeof updates.enabled !== "undefined") registeredPrompt.enabled = updates.enabled
|
||||
this.sendPromptListChanged()
|
||||
},
|
||||
};
|
||||
this._registeredPrompts[name] = registeredPrompt;
|
||||
|
||||
this.setPromptRequestHandlers();
|
||||
this.sendPromptListChanged()
|
||||
|
||||
return registeredPrompt
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks if the server is connected to a transport.
|
||||
* @returns True if the server is connected
|
||||
*/
|
||||
isConnected() {
|
||||
return this.server.transport !== undefined
|
||||
}
|
||||
|
||||
/**
|
||||
* Sends a resource list changed event to the client, if connected.
|
||||
*/
|
||||
sendResourceListChanged() {
|
||||
if (this.isConnected()) {
|
||||
this.server.sendResourceListChanged();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Sends a tool list changed event to the client, if connected.
|
||||
*/
|
||||
sendToolListChanged() {
|
||||
if (this.isConnected()) {
|
||||
this.server.sendToolListChanged();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Sends a prompt list changed event to the client, if connected.
|
||||
*/
|
||||
sendPromptListChanged() {
|
||||
if (this.isConnected()) {
|
||||
this.server.sendPromptListChanged();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -694,14 +846,19 @@ export type ToolCallback<Args extends undefined | ZodRawShape = undefined> =
|
|||
Args extends ZodRawShape
|
||||
? (
|
||||
args: z.objectOutputType<Args, ZodTypeAny>,
|
||||
extra: RequestHandlerExtra,
|
||||
extra: RequestHandlerExtra<ServerRequest, ServerNotification>,
|
||||
) => CallToolResult | Promise<CallToolResult>
|
||||
: (extra: RequestHandlerExtra) => CallToolResult | Promise<CallToolResult>;
|
||||
: (extra: RequestHandlerExtra<ServerRequest, ServerNotification>) => CallToolResult | Promise<CallToolResult>;
|
||||
|
||||
type RegisteredTool = {
|
||||
export type RegisteredTool = {
|
||||
description?: string;
|
||||
inputSchema?: AnyZodObject;
|
||||
callback: ToolCallback<undefined | ZodRawShape>;
|
||||
enabled: boolean;
|
||||
enable(): void;
|
||||
disable(): void;
|
||||
update<Args extends ZodRawShape>(updates: { name?: string | null, description?: string, paramsSchema?: Args, callback?: ToolCallback<Args>, enabled?: boolean }): void
|
||||
remove(): void
|
||||
};
|
||||
|
||||
const EMPTY_OBJECT_JSON_SCHEMA = {
|
||||
|
|
@ -717,7 +874,7 @@ export type ResourceMetadata = Omit<Resource, "uri" | "name">;
|
|||
* Callback to list all resources matching a given template.
|
||||
*/
|
||||
export type ListResourcesCallback = (
|
||||
extra: RequestHandlerExtra,
|
||||
extra: RequestHandlerExtra<ServerRequest, ServerNotification>,
|
||||
) => ListResourcesResult | Promise<ListResourcesResult>;
|
||||
|
||||
/**
|
||||
|
|
@ -725,13 +882,18 @@ export type ListResourcesCallback = (
|
|||
*/
|
||||
export type ReadResourceCallback = (
|
||||
uri: URL,
|
||||
extra: RequestHandlerExtra,
|
||||
extra: RequestHandlerExtra<ServerRequest, ServerNotification>,
|
||||
) => ReadResourceResult | Promise<ReadResourceResult>;
|
||||
|
||||
type RegisteredResource = {
|
||||
export type RegisteredResource = {
|
||||
name: string;
|
||||
metadata?: ResourceMetadata;
|
||||
readCallback: ReadResourceCallback;
|
||||
enabled: boolean;
|
||||
enable(): void;
|
||||
disable(): void;
|
||||
update(updates: { name?: string, uri?: string | null, metadata?: ResourceMetadata, callback?: ReadResourceCallback, enabled?: boolean }): void
|
||||
remove(): void
|
||||
};
|
||||
|
||||
/**
|
||||
|
|
@ -740,13 +902,18 @@ type RegisteredResource = {
|
|||
export type ReadResourceTemplateCallback = (
|
||||
uri: URL,
|
||||
variables: Variables,
|
||||
extra: RequestHandlerExtra,
|
||||
extra: RequestHandlerExtra<ServerRequest, ServerNotification>,
|
||||
) => ReadResourceResult | Promise<ReadResourceResult>;
|
||||
|
||||
type RegisteredResourceTemplate = {
|
||||
export type RegisteredResourceTemplate = {
|
||||
resourceTemplate: ResourceTemplate;
|
||||
metadata?: ResourceMetadata;
|
||||
readCallback: ReadResourceTemplateCallback;
|
||||
enabled: boolean;
|
||||
enable(): void;
|
||||
disable(): void;
|
||||
update(updates: { name?: string | null, template?: ResourceTemplate, metadata?: ResourceMetadata, callback?: ReadResourceTemplateCallback, enabled?: boolean }): void
|
||||
remove(): void
|
||||
};
|
||||
|
||||
type PromptArgsRawShape = {
|
||||
|
|
@ -760,14 +927,19 @@ export type PromptCallback<
|
|||
> = Args extends PromptArgsRawShape
|
||||
? (
|
||||
args: z.objectOutputType<Args, ZodTypeAny>,
|
||||
extra: RequestHandlerExtra,
|
||||
extra: RequestHandlerExtra<ServerRequest, ServerNotification>,
|
||||
) => GetPromptResult | Promise<GetPromptResult>
|
||||
: (extra: RequestHandlerExtra) => GetPromptResult | Promise<GetPromptResult>;
|
||||
: (extra: RequestHandlerExtra<ServerRequest, ServerNotification>) => GetPromptResult | Promise<GetPromptResult>;
|
||||
|
||||
type RegisteredPrompt = {
|
||||
export type RegisteredPrompt = {
|
||||
description?: string;
|
||||
argsSchema?: ZodObject<PromptArgsRawShape>;
|
||||
callback: PromptCallback<undefined | PromptArgsRawShape>;
|
||||
enabled: boolean;
|
||||
enable(): void;
|
||||
disable(): void;
|
||||
update<Args extends PromptArgsRawShape>(updates: { name?: string | null, description?: string, argsSchema?: Args, callback?: PromptCallback<Args>, enabled?: boolean }): void
|
||||
remove(): void
|
||||
};
|
||||
|
||||
function promptArgumentsFromSchema(
|
||||
|
|
|
|||
109
src/server/sse.test.ts
Normal file
109
src/server/sse.test.ts
Normal file
|
|
@ -0,0 +1,109 @@
|
|||
import http from 'http';
|
||||
import { jest } from '@jest/globals';
|
||||
import { SSEServerTransport } from './sse.js';
|
||||
|
||||
const createMockResponse = () => {
|
||||
const res = {
|
||||
writeHead: jest.fn<http.ServerResponse['writeHead']>(),
|
||||
write: jest.fn<http.ServerResponse['write']>().mockReturnValue(true),
|
||||
on: jest.fn<http.ServerResponse['on']>(),
|
||||
};
|
||||
res.writeHead.mockReturnThis();
|
||||
res.on.mockReturnThis();
|
||||
|
||||
return res as unknown as http.ServerResponse;
|
||||
};
|
||||
|
||||
describe('SSEServerTransport', () => {
|
||||
describe('start method', () => {
|
||||
it('should correctly append sessionId to a simple relative endpoint', async () => {
|
||||
const mockRes = createMockResponse();
|
||||
const endpoint = '/messages';
|
||||
const transport = new SSEServerTransport(endpoint, mockRes);
|
||||
const expectedSessionId = transport.sessionId;
|
||||
|
||||
await transport.start();
|
||||
|
||||
expect(mockRes.writeHead).toHaveBeenCalledWith(200, expect.any(Object));
|
||||
expect(mockRes.write).toHaveBeenCalledTimes(1);
|
||||
expect(mockRes.write).toHaveBeenCalledWith(
|
||||
`event: endpoint\ndata: /messages?sessionId=${expectedSessionId}\n\n`
|
||||
);
|
||||
});
|
||||
|
||||
it('should correctly append sessionId to an endpoint with existing query parameters', async () => {
|
||||
const mockRes = createMockResponse();
|
||||
const endpoint = '/messages?foo=bar&baz=qux';
|
||||
const transport = new SSEServerTransport(endpoint, mockRes);
|
||||
const expectedSessionId = transport.sessionId;
|
||||
|
||||
await transport.start();
|
||||
|
||||
expect(mockRes.writeHead).toHaveBeenCalledWith(200, expect.any(Object));
|
||||
expect(mockRes.write).toHaveBeenCalledTimes(1);
|
||||
expect(mockRes.write).toHaveBeenCalledWith(
|
||||
`event: endpoint\ndata: /messages?foo=bar&baz=qux&sessionId=${expectedSessionId}\n\n`
|
||||
);
|
||||
});
|
||||
|
||||
it('should correctly append sessionId to an endpoint with a hash fragment', async () => {
|
||||
const mockRes = createMockResponse();
|
||||
const endpoint = '/messages#section1';
|
||||
const transport = new SSEServerTransport(endpoint, mockRes);
|
||||
const expectedSessionId = transport.sessionId;
|
||||
|
||||
await transport.start();
|
||||
|
||||
expect(mockRes.writeHead).toHaveBeenCalledWith(200, expect.any(Object));
|
||||
expect(mockRes.write).toHaveBeenCalledTimes(1);
|
||||
expect(mockRes.write).toHaveBeenCalledWith(
|
||||
`event: endpoint\ndata: /messages?sessionId=${expectedSessionId}#section1\n\n`
|
||||
);
|
||||
});
|
||||
|
||||
it('should correctly append sessionId to an endpoint with query parameters and a hash fragment', async () => {
|
||||
const mockRes = createMockResponse();
|
||||
const endpoint = '/messages?key=value#section2';
|
||||
const transport = new SSEServerTransport(endpoint, mockRes);
|
||||
const expectedSessionId = transport.sessionId;
|
||||
|
||||
await transport.start();
|
||||
|
||||
expect(mockRes.writeHead).toHaveBeenCalledWith(200, expect.any(Object));
|
||||
expect(mockRes.write).toHaveBeenCalledTimes(1);
|
||||
expect(mockRes.write).toHaveBeenCalledWith(
|
||||
`event: endpoint\ndata: /messages?key=value&sessionId=${expectedSessionId}#section2\n\n`
|
||||
);
|
||||
});
|
||||
|
||||
it('should correctly handle the root path endpoint "/"', async () => {
|
||||
const mockRes = createMockResponse();
|
||||
const endpoint = '/';
|
||||
const transport = new SSEServerTransport(endpoint, mockRes);
|
||||
const expectedSessionId = transport.sessionId;
|
||||
|
||||
await transport.start();
|
||||
|
||||
expect(mockRes.writeHead).toHaveBeenCalledWith(200, expect.any(Object));
|
||||
expect(mockRes.write).toHaveBeenCalledTimes(1);
|
||||
expect(mockRes.write).toHaveBeenCalledWith(
|
||||
`event: endpoint\ndata: /?sessionId=${expectedSessionId}\n\n`
|
||||
);
|
||||
});
|
||||
|
||||
it('should correctly handle an empty string endpoint ""', async () => {
|
||||
const mockRes = createMockResponse();
|
||||
const endpoint = '';
|
||||
const transport = new SSEServerTransport(endpoint, mockRes);
|
||||
const expectedSessionId = transport.sessionId;
|
||||
|
||||
await transport.start();
|
||||
|
||||
expect(mockRes.writeHead).toHaveBeenCalledWith(200, expect.any(Object));
|
||||
expect(mockRes.write).toHaveBeenCalledTimes(1);
|
||||
expect(mockRes.write).toHaveBeenCalledWith(
|
||||
`event: endpoint\ndata: /?sessionId=${expectedSessionId}\n\n`
|
||||
);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
@ -5,6 +5,7 @@ import { JSONRPCMessage, JSONRPCMessageSchema } from "../types.js";
|
|||
import getRawBody from "raw-body";
|
||||
import contentType from "content-type";
|
||||
import { AuthInfo } from "./auth/types.js";
|
||||
import { URL } from 'url';
|
||||
|
||||
const MAXIMUM_MESSAGE_SIZE = "4mb";
|
||||
|
||||
|
|
@ -50,8 +51,17 @@ export class SSEServerTransport implements Transport {
|
|||
});
|
||||
|
||||
// Send the endpoint event
|
||||
// Use a dummy base URL because this._endpoint is relative.
|
||||
// This allows using URL/URLSearchParams for robust parameter handling.
|
||||
const dummyBase = 'http://localhost'; // Any valid base works
|
||||
const endpointUrl = new URL(this._endpoint, dummyBase);
|
||||
endpointUrl.searchParams.set('sessionId', this._sessionId);
|
||||
|
||||
// Reconstruct the relative URL string (pathname + search + hash)
|
||||
const relativeUrlWithSession = endpointUrl.pathname + endpointUrl.search + endpointUrl.hash;
|
||||
|
||||
this.res.write(
|
||||
`event: endpoint\ndata: ${encodeURI(this._endpoint)}?sessionId=${this._sessionId}\n\n`,
|
||||
`event: endpoint\ndata: ${relativeUrlWithSession}\n\n`,
|
||||
);
|
||||
|
||||
this._sseResponse = this.res;
|
||||
|
|
|
|||
1590
src/server/streamableHttp.test.ts
Normal file
1590
src/server/streamableHttp.test.ts
Normal file
File diff suppressed because it is too large
Load diff
523
src/server/streamableHttp.ts
Normal file
523
src/server/streamableHttp.ts
Normal file
|
|
@ -0,0 +1,523 @@
|
|||
import { IncomingMessage, ServerResponse } from "node:http";
|
||||
import { Transport } from "../shared/transport.js";
|
||||
import { JSONRPCMessage, JSONRPCMessageSchema, RequestId } from "../types.js";
|
||||
import getRawBody from "raw-body";
|
||||
import contentType from "content-type";
|
||||
|
||||
const MAXIMUM_MESSAGE_SIZE = "4mb";
|
||||
|
||||
/**
|
||||
* Configuration options for StreamableHTTPServerTransport
|
||||
*/
|
||||
export interface StreamableHTTPServerTransportOptions {
|
||||
/**
|
||||
* Function that generates a session ID for the transport.
|
||||
* The session ID SHOULD be globally unique and cryptographically secure (e.g., a securely generated UUID, a JWT, or a cryptographic hash)
|
||||
*
|
||||
* Return undefined to disable session management.
|
||||
*/
|
||||
sessionIdGenerator: () => string | undefined;
|
||||
|
||||
/**
|
||||
* If true, the server will return JSON responses instead of starting an SSE stream.
|
||||
* This can be useful for simple request/response scenarios without streaming.
|
||||
* Default is false (SSE streams are preferred).
|
||||
*/
|
||||
enableJsonResponse?: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* Server transport for Streamable HTTP: this implements the MCP Streamable HTTP transport specification.
|
||||
* It supports both SSE streaming and direct HTTP responses.
|
||||
*
|
||||
* Usage example:
|
||||
*
|
||||
* ```typescript
|
||||
* // Stateful mode - server sets the session ID
|
||||
* const statefulTransport = new StreamableHTTPServerTransport({
|
||||
* sessionId: randomUUID(),
|
||||
* });
|
||||
*
|
||||
* // Stateless mode - explicitly set session ID to undefined
|
||||
* const statelessTransport = new StreamableHTTPServerTransport({
|
||||
* sessionId: undefined,
|
||||
* });
|
||||
*
|
||||
* // Using with pre-parsed request body
|
||||
* app.post('/mcp', (req, res) => {
|
||||
* transport.handleRequest(req, res, req.body);
|
||||
* });
|
||||
* ```
|
||||
*
|
||||
* In stateful mode:
|
||||
* - Session ID is generated and included in response headers
|
||||
* - Session ID is always included in initialization responses
|
||||
* - Requests with invalid session IDs are rejected with 404 Not Found
|
||||
* - Non-initialization requests without a session ID are rejected with 400 Bad Request
|
||||
* - State is maintained in-memory (connections, message history)
|
||||
*
|
||||
* In stateless mode:
|
||||
* - Session ID is only included in initialization responses
|
||||
* - No session validation is performed
|
||||
*/
|
||||
export class StreamableHTTPServerTransport implements Transport {
|
||||
// when sessionId is not set (undefined), it means the transport is in stateless mode
|
||||
private sessionIdGenerator: () => string | undefined;
|
||||
private _started: boolean = false;
|
||||
private _responseMapping: Map<RequestId, ServerResponse> = new Map();
|
||||
private _requestResponseMap: Map<RequestId, JSONRPCMessage> = new Map();
|
||||
private _initialized: boolean = false;
|
||||
private _enableJsonResponse: boolean = false;
|
||||
private _standaloneSSE: ServerResponse | undefined;
|
||||
|
||||
|
||||
sessionId?: string | undefined;
|
||||
onclose?: () => void;
|
||||
onerror?: (error: Error) => void;
|
||||
onmessage?: (message: JSONRPCMessage) => void;
|
||||
|
||||
constructor(options: StreamableHTTPServerTransportOptions) {
|
||||
this.sessionIdGenerator = options.sessionIdGenerator;
|
||||
this._enableJsonResponse = options.enableJsonResponse ?? false;
|
||||
}
|
||||
|
||||
/**
|
||||
* Starts the transport. This is required by the Transport interface but is a no-op
|
||||
* for the Streamable HTTP transport as connections are managed per-request.
|
||||
*/
|
||||
async start(): Promise<void> {
|
||||
if (this._started) {
|
||||
throw new Error("Transport already started");
|
||||
}
|
||||
this._started = true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Handles an incoming HTTP request, whether GET or POST
|
||||
*/
|
||||
async handleRequest(req: IncomingMessage, res: ServerResponse, parsedBody?: unknown): Promise<void> {
|
||||
if (req.method === "POST") {
|
||||
await this.handlePostRequest(req, res, parsedBody);
|
||||
} else if (req.method === "GET") {
|
||||
await this.handleGetRequest(req, res);
|
||||
} else if (req.method === "DELETE") {
|
||||
await this.handleDeleteRequest(req, res);
|
||||
} else {
|
||||
await this.handleUnsupportedRequest(res);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Handles GET requests for SSE stream
|
||||
*/
|
||||
private async handleGetRequest(req: IncomingMessage, res: ServerResponse): Promise<void> {
|
||||
// The client MUST include an Accept header, listing text/event-stream as a supported content type.
|
||||
const acceptHeader = req.headers.accept;
|
||||
if (!acceptHeader?.includes("text/event-stream")) {
|
||||
res.writeHead(406).end(JSON.stringify({
|
||||
jsonrpc: "2.0",
|
||||
error: {
|
||||
code: -32000,
|
||||
message: "Not Acceptable: Client must accept text/event-stream"
|
||||
},
|
||||
id: null
|
||||
}));
|
||||
return;
|
||||
}
|
||||
|
||||
// If an Mcp-Session-Id is returned by the server during initialization,
|
||||
// clients using the Streamable HTTP transport MUST include it
|
||||
// in the Mcp-Session-Id header on all of their subsequent HTTP requests.
|
||||
if (!this.validateSession(req, res)) {
|
||||
return;
|
||||
}
|
||||
|
||||
// The server MUST either return Content-Type: text/event-stream in response to this HTTP GET,
|
||||
// or else return HTTP 405 Method Not Allowed
|
||||
const headers: Record<string, string> = {
|
||||
"Content-Type": "text/event-stream",
|
||||
"Cache-Control": "no-cache, no-transform",
|
||||
Connection: "keep-alive",
|
||||
};
|
||||
|
||||
// After initialization, always include the session ID if we have one
|
||||
if (this.sessionId !== undefined) {
|
||||
headers["mcp-session-id"] = this.sessionId;
|
||||
}
|
||||
// The server MAY include a Last-Event-ID header in the response to this HTTP GET.
|
||||
// Resumability will be supported in the future
|
||||
|
||||
// Check if there's already an active standalone SSE stream for this session
|
||||
|
||||
if (this._standaloneSSE !== undefined) {
|
||||
// Only one GET SSE stream is allowed per session
|
||||
res.writeHead(409).end(JSON.stringify({
|
||||
jsonrpc: "2.0",
|
||||
error: {
|
||||
code: -32000,
|
||||
message: "Conflict: Only one SSE stream is allowed per session"
|
||||
},
|
||||
id: null
|
||||
}));
|
||||
return;
|
||||
}
|
||||
// We need to send headers immediately as message will arrive much later,
|
||||
// otherwise the client will just wait for the first message
|
||||
res.writeHead(200, headers).flushHeaders();
|
||||
|
||||
// Assing the response to the standalone SSE stream
|
||||
this._standaloneSSE = res;
|
||||
|
||||
// Set up close handler for client disconnects
|
||||
res.on("close", () => {
|
||||
this._standaloneSSE = undefined;
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Handles unsupported requests (PUT, PATCH, etc.)
|
||||
*/
|
||||
private async handleUnsupportedRequest(res: ServerResponse): Promise<void> {
|
||||
res.writeHead(405, {
|
||||
"Allow": "GET, POST, DELETE"
|
||||
}).end(JSON.stringify({
|
||||
jsonrpc: "2.0",
|
||||
error: {
|
||||
code: -32000,
|
||||
message: "Method not allowed."
|
||||
},
|
||||
id: null
|
||||
}));
|
||||
}
|
||||
|
||||
/**
|
||||
* Handles POST requests containing JSON-RPC messages
|
||||
*/
|
||||
private async handlePostRequest(req: IncomingMessage, res: ServerResponse, parsedBody?: unknown): Promise<void> {
|
||||
try {
|
||||
// Validate the Accept header
|
||||
const acceptHeader = req.headers.accept;
|
||||
// The client MUST include an Accept header, listing both application/json and text/event-stream as supported content types.
|
||||
if (!acceptHeader?.includes("application/json") || !acceptHeader.includes("text/event-stream")) {
|
||||
res.writeHead(406).end(JSON.stringify({
|
||||
jsonrpc: "2.0",
|
||||
error: {
|
||||
code: -32000,
|
||||
message: "Not Acceptable: Client must accept both application/json and text/event-stream"
|
||||
},
|
||||
id: null
|
||||
}));
|
||||
return;
|
||||
}
|
||||
|
||||
const ct = req.headers["content-type"];
|
||||
if (!ct || !ct.includes("application/json")) {
|
||||
res.writeHead(415).end(JSON.stringify({
|
||||
jsonrpc: "2.0",
|
||||
error: {
|
||||
code: -32000,
|
||||
message: "Unsupported Media Type: Content-Type must be application/json"
|
||||
},
|
||||
id: null
|
||||
}));
|
||||
return;
|
||||
}
|
||||
|
||||
let rawMessage;
|
||||
if (parsedBody !== undefined) {
|
||||
rawMessage = parsedBody;
|
||||
} else {
|
||||
const parsedCt = contentType.parse(ct);
|
||||
const body = await getRawBody(req, {
|
||||
limit: MAXIMUM_MESSAGE_SIZE,
|
||||
encoding: parsedCt.parameters.charset ?? "utf-8",
|
||||
});
|
||||
rawMessage = JSON.parse(body.toString());
|
||||
}
|
||||
|
||||
let messages: JSONRPCMessage[];
|
||||
|
||||
// handle batch and single messages
|
||||
if (Array.isArray(rawMessage)) {
|
||||
messages = rawMessage.map(msg => JSONRPCMessageSchema.parse(msg));
|
||||
} else {
|
||||
messages = [JSONRPCMessageSchema.parse(rawMessage)];
|
||||
}
|
||||
|
||||
// Check if this is an initialization request
|
||||
// https://spec.modelcontextprotocol.io/specification/2025-03-26/basic/lifecycle/
|
||||
const isInitializationRequest = messages.some(
|
||||
msg => 'method' in msg && msg.method === 'initialize'
|
||||
);
|
||||
if (isInitializationRequest) {
|
||||
if (this._initialized) {
|
||||
res.writeHead(400).end(JSON.stringify({
|
||||
jsonrpc: "2.0",
|
||||
error: {
|
||||
code: -32600,
|
||||
message: "Invalid Request: Server already initialized"
|
||||
},
|
||||
id: null
|
||||
}));
|
||||
return;
|
||||
}
|
||||
if (messages.length > 1) {
|
||||
res.writeHead(400).end(JSON.stringify({
|
||||
jsonrpc: "2.0",
|
||||
error: {
|
||||
code: -32600,
|
||||
message: "Invalid Request: Only one initialization request is allowed"
|
||||
},
|
||||
id: null
|
||||
}));
|
||||
return;
|
||||
}
|
||||
this.sessionId = this.sessionIdGenerator();
|
||||
this._initialized = true;
|
||||
|
||||
}
|
||||
// If an Mcp-Session-Id is returned by the server during initialization,
|
||||
// clients using the Streamable HTTP transport MUST include it
|
||||
// in the Mcp-Session-Id header on all of their subsequent HTTP requests.
|
||||
if (!isInitializationRequest && !this.validateSession(req, res)) {
|
||||
return;
|
||||
}
|
||||
|
||||
|
||||
// check if it contains requests
|
||||
const hasRequests = messages.some(msg => 'method' in msg && 'id' in msg);
|
||||
const hasOnlyNotificationsOrResponses = messages.every(msg =>
|
||||
('method' in msg && !('id' in msg)) || ('result' in msg || 'error' in msg));
|
||||
|
||||
if (hasOnlyNotificationsOrResponses) {
|
||||
// if it only contains notifications or responses, return 202
|
||||
res.writeHead(202).end();
|
||||
|
||||
// handle each message
|
||||
for (const message of messages) {
|
||||
this.onmessage?.(message);
|
||||
}
|
||||
} else if (hasRequests) {
|
||||
// The default behavior is to use SSE streaming
|
||||
// but in some cases server will return JSON responses
|
||||
if (!this._enableJsonResponse) {
|
||||
const headers: Record<string, string> = {
|
||||
"Content-Type": "text/event-stream",
|
||||
"Cache-Control": "no-cache",
|
||||
Connection: "keep-alive",
|
||||
};
|
||||
|
||||
// After initialization, always include the session ID if we have one
|
||||
if (this.sessionId !== undefined) {
|
||||
headers["mcp-session-id"] = this.sessionId;
|
||||
}
|
||||
|
||||
res.writeHead(200, headers);
|
||||
}
|
||||
// Store the response for this request to send messages back through this connection
|
||||
// We need to track by request ID to maintain the connection
|
||||
for (const message of messages) {
|
||||
if ('method' in message && 'id' in message) {
|
||||
this._responseMapping.set(message.id, res);
|
||||
}
|
||||
}
|
||||
|
||||
// Set up close handler for client disconnects
|
||||
res.on("close", () => {
|
||||
// Remove all entries that reference this response
|
||||
for (const [id, storedRes] of this._responseMapping.entries()) {
|
||||
if (storedRes === res) {
|
||||
this._responseMapping.delete(id);
|
||||
this._requestResponseMap.delete(id);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
// handle each message
|
||||
for (const message of messages) {
|
||||
this.onmessage?.(message);
|
||||
}
|
||||
// The server SHOULD NOT close the SSE stream before sending all JSON-RPC responses
|
||||
// This will be handled by the send() method when responses are ready
|
||||
}
|
||||
} catch (error) {
|
||||
// return JSON-RPC formatted error
|
||||
res.writeHead(400).end(JSON.stringify({
|
||||
jsonrpc: "2.0",
|
||||
error: {
|
||||
code: -32700,
|
||||
message: "Parse error",
|
||||
data: String(error)
|
||||
},
|
||||
id: null
|
||||
}));
|
||||
this.onerror?.(error as Error);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Handles DELETE requests to terminate sessions
|
||||
*/
|
||||
private async handleDeleteRequest(req: IncomingMessage, res: ServerResponse): Promise<void> {
|
||||
if (!this.validateSession(req, res)) {
|
||||
return;
|
||||
}
|
||||
await this.close();
|
||||
res.writeHead(200).end();
|
||||
}
|
||||
|
||||
/**
|
||||
* Validates session ID for non-initialization requests
|
||||
* Returns true if the session is valid, false otherwise
|
||||
*/
|
||||
private validateSession(req: IncomingMessage, res: ServerResponse): boolean {
|
||||
if (!this._initialized) {
|
||||
// If the server has not been initialized yet, reject all requests
|
||||
res.writeHead(400).end(JSON.stringify({
|
||||
jsonrpc: "2.0",
|
||||
error: {
|
||||
code: -32000,
|
||||
message: "Bad Request: Server not initialized"
|
||||
},
|
||||
id: null
|
||||
}));
|
||||
return false;
|
||||
}
|
||||
if (this.sessionId === undefined) {
|
||||
// If the session ID is not set, the session management is disabled
|
||||
// and we don't need to validate the session ID
|
||||
return true;
|
||||
}
|
||||
const sessionId = req.headers["mcp-session-id"];
|
||||
|
||||
if (!sessionId) {
|
||||
// Non-initialization requests without a session ID should return 400 Bad Request
|
||||
res.writeHead(400).end(JSON.stringify({
|
||||
jsonrpc: "2.0",
|
||||
error: {
|
||||
code: -32000,
|
||||
message: "Bad Request: Mcp-Session-Id header is required"
|
||||
},
|
||||
id: null
|
||||
}));
|
||||
return false;
|
||||
} else if (Array.isArray(sessionId)) {
|
||||
res.writeHead(400).end(JSON.stringify({
|
||||
jsonrpc: "2.0",
|
||||
error: {
|
||||
code: -32000,
|
||||
message: "Bad Request: Mcp-Session-Id header must be a single value"
|
||||
},
|
||||
id: null
|
||||
}));
|
||||
return false;
|
||||
}
|
||||
else if (sessionId !== this.sessionId) {
|
||||
// Reject requests with invalid session ID with 404 Not Found
|
||||
res.writeHead(404).end(JSON.stringify({
|
||||
jsonrpc: "2.0",
|
||||
error: {
|
||||
code: -32001,
|
||||
message: "Session not found"
|
||||
},
|
||||
id: null
|
||||
}));
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
|
||||
async close(): Promise<void> {
|
||||
// Close all SSE connections
|
||||
this._responseMapping.forEach((response) => {
|
||||
response.end();
|
||||
});
|
||||
this._responseMapping.clear();
|
||||
|
||||
// Clear any pending responses
|
||||
this._requestResponseMap.clear();
|
||||
|
||||
this.onclose?.();
|
||||
}
|
||||
|
||||
async send(message: JSONRPCMessage, options?: { relatedRequestId?: RequestId }): Promise<void> {
|
||||
let requestId = options?.relatedRequestId;
|
||||
if ('result' in message || 'error' in message) {
|
||||
// If the message is a response, use the request ID from the message
|
||||
requestId = message.id;
|
||||
}
|
||||
|
||||
// Check if this message should be sent on the standalone SSE stream (no request ID)
|
||||
// Ignore notifications from tools (which have relatedRequestId set)
|
||||
// Those will be sent via dedicated response SSE streams
|
||||
if (requestId === undefined) {
|
||||
// For standalone SSE streams, we can only send requests and notifications
|
||||
if ('result' in message || 'error' in message) {
|
||||
throw new Error("Cannot send a response on a standalone SSE stream unless resuming a previous client request");
|
||||
}
|
||||
|
||||
if (this._standaloneSSE === undefined) {
|
||||
// The spec says the server MAY send messages on the stream, so it's ok to discard if no stream
|
||||
return;
|
||||
}
|
||||
|
||||
// Send the message to the standalone SSE stream
|
||||
this._standaloneSSE.write(`event: message\ndata: ${JSON.stringify(message)}\n\n`);
|
||||
return;
|
||||
}
|
||||
|
||||
// Get the response for this request
|
||||
const response = this._responseMapping.get(requestId);
|
||||
if (!response) {
|
||||
throw new Error(`No connection established for request ID: ${String(requestId)}`);
|
||||
}
|
||||
|
||||
if (!this._enableJsonResponse) {
|
||||
response.write(`event: message\ndata: ${JSON.stringify(message)}\n\n`);
|
||||
}
|
||||
if ('result' in message || 'error' in message) {
|
||||
this._requestResponseMap.set(requestId, message);
|
||||
|
||||
// Get all request IDs that share the same request response object
|
||||
const relatedIds = Array.from(this._responseMapping.entries())
|
||||
.filter(([_, res]) => res === response)
|
||||
.map(([id]) => id);
|
||||
|
||||
// Check if we have responses for all requests using this connection
|
||||
const allResponsesReady = relatedIds.every(id => this._requestResponseMap.has(id));
|
||||
|
||||
if (allResponsesReady) {
|
||||
if (this._enableJsonResponse) {
|
||||
// All responses ready, send as JSON
|
||||
const headers: Record<string, string> = {
|
||||
'Content-Type': 'application/json',
|
||||
};
|
||||
if (this.sessionId !== undefined) {
|
||||
headers['mcp-session-id'] = this.sessionId;
|
||||
}
|
||||
|
||||
const responses = relatedIds
|
||||
.map(id => this._requestResponseMap.get(id)!);
|
||||
|
||||
response.writeHead(200, headers);
|
||||
if (responses.length === 1) {
|
||||
response.end(JSON.stringify(responses[0]));
|
||||
} else {
|
||||
response.end(JSON.stringify(responses));
|
||||
}
|
||||
} else {
|
||||
// End the SSE stream
|
||||
response.end();
|
||||
}
|
||||
// Clean up
|
||||
for (const id of relatedIds) {
|
||||
this._requestResponseMap.delete(id);
|
||||
this._responseMapping.delete(id);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Loading…
Reference in a new issue