diff --git a/src/client/streamableHttp.ts b/src/client/streamableHttp.ts index 077b0f15..0510fe84 100644 --- a/src/client/streamableHttp.ts +++ b/src/client/streamableHttp.ts @@ -1,5 +1,5 @@ import { Transport } from "../shared/transport.js"; -import { isJSONRPCNotification, isJSONRPCRequest, isJSONRPCResponse, JSONRPCMessage, JSONRPCMessageSchema } from "../types.js"; +import { isInitializedNotification, isJSONRPCNotification, isJSONRPCRequest, isJSONRPCResponse, JSONRPCMessage, JSONRPCMessageSchema } from "../types.js"; import { auth, AuthResult, OAuthClientProvider, UnauthorizedError } from "./auth.js"; import { EventSourceParserStream } from "eventsource-parser/stream"; @@ -420,7 +420,7 @@ export class StreamableHTTPClientTransport implements Transport { if (response.status === 202) { // if the accepted notification is initialized, we start the SSE stream // if it's supported by the server - if (isJSONRPCNotification(message) && message.method === "notifications/initialized") { + if (isInitializedNotification(message)) { // Start without a lastEventId since this is a fresh connection this._startOrAuthSse({ resumptionToken: undefined }).catch(err => this.onerror?.(err)); } diff --git a/src/examples/server/jsonResponseStreamableHttp.ts b/src/examples/server/jsonResponseStreamableHttp.ts index 34ab65d1..9be63f52 100644 --- a/src/examples/server/jsonResponseStreamableHttp.ts +++ b/src/examples/server/jsonResponseStreamableHttp.ts @@ -3,7 +3,7 @@ import { randomUUID } from 'node:crypto'; import { McpServer } from '../../server/mcp.js'; import { StreamableHTTPServerTransport } from '../../server/streamableHttp.js'; import { z } from 'zod'; -import { CallToolResult } from '../../types.js'; +import { CallToolResult, isInitializeRequest } from '../../types.js'; // Create an MCP server with implementation details const server = new McpServer({ @@ -145,14 +145,6 @@ app.get('/mcp', async (req: Request, res: Response) => { res.status(405).set('Allow', 'POST').send('Method Not Allowed'); }); -// Helper function to detect initialize requests -function isInitializeRequest(body: unknown): boolean { - if (Array.isArray(body)) { - return body.some(msg => typeof msg === 'object' && msg !== null && 'method' in msg && msg.method === 'initialize'); - } - return typeof body === 'object' && body !== null && 'method' in body && body.method === 'initialize'; -} - // Start the server const PORT = 3000; app.listen(PORT, () => { diff --git a/src/examples/server/simpleStreamableHttp.ts b/src/examples/server/simpleStreamableHttp.ts index 98333730..eb28bf4c 100644 --- a/src/examples/server/simpleStreamableHttp.ts +++ b/src/examples/server/simpleStreamableHttp.ts @@ -3,7 +3,7 @@ import { randomUUID } from 'node:crypto'; import { McpServer } from '../../server/mcp.js'; import { EventStore, StreamableHTTPServerTransport } from '../../server/streamableHttp.js'; import { z } from 'zod'; -import { CallToolResult, GetPromptResult, JSONRPCMessage, ReadResourceResult } from '../../types.js'; +import { CallToolResult, GetPromptResult, isInitializeRequest, JSONRPCMessage, ReadResourceResult } from '../../types.js'; // Create a simple in-memory EventStore for resumability class InMemoryEventStore implements EventStore { @@ -36,7 +36,7 @@ class InMemoryEventStore implements EventStore { * Replays events that occurred after a specific event ID * Implements EventStore.replayEventsAfter */ - async replayEventsAfter(lastEventId: string, + async replayEventsAfter(lastEventId: string, { send }: { send: (eventId: string, message: JSONRPCMessage) => Promise } ): Promise { if (!lastEventId || !this.events.has(lastEventId)) { @@ -312,14 +312,6 @@ app.get('/mcp', async (req: Request, res: Response) => { await transport.handleRequest(req, res); }); -// Helper function to detect initialize requests -function isInitializeRequest(body: unknown): boolean { - if (Array.isArray(body)) { - return body.some(msg => typeof msg === 'object' && msg !== null && 'method' in msg && msg.method === 'initialize'); - } - return typeof body === 'object' && body !== null && 'method' in body && body.method === 'initialize'; -} - // Start the server const PORT = 3000; app.listen(PORT, () => { diff --git a/src/examples/server/standaloneSseWithGetStreamableHttp.ts b/src/examples/server/standaloneSseWithGetStreamableHttp.ts index f9d3696b..5f96ca82 100644 --- a/src/examples/server/standaloneSseWithGetStreamableHttp.ts +++ b/src/examples/server/standaloneSseWithGetStreamableHttp.ts @@ -2,7 +2,7 @@ import express, { Request, Response } from 'express'; import { randomUUID } from 'node:crypto'; import { McpServer } from '../../server/mcp.js'; import { StreamableHTTPServerTransport } from '../../server/streamableHttp.js'; -import { ReadResourceResult } from '../../types.js'; +import { isInitializeRequest, ReadResourceResult } from '../../types.js'; // Create an MCP server with implementation details const server = new McpServer({ @@ -107,13 +107,6 @@ app.get('/mcp', async (req: Request, res: Response) => { await transport.handleRequest(req, res); }); -// Helper function to detect initialize requests -function isInitializeRequest(body: unknown): boolean { - if (Array.isArray(body)) { - return body.some(msg => typeof msg === 'object' && msg !== null && 'method' in msg && msg.method === 'initialize'); - } - return typeof body === 'object' && body !== null && 'method' in body && body.method === 'initialize'; -} // Start the server const PORT = 3000; diff --git a/src/types.ts b/src/types.ts index db6bf125..8ac41372 100644 --- a/src/types.ts +++ b/src/types.ts @@ -248,6 +248,10 @@ export const InitializeRequestSchema = RequestSchema.extend({ }), }); +export const isInitializeRequest = (value: unknown): value is InitializeRequest => + InitializeRequestSchema.safeParse(value).success; + + /** * Capabilities that a server may support. Known capabilities are defined here, in this schema, but this is not a closed set: any server can define its own, additional capabilities. */ @@ -337,6 +341,9 @@ export const InitializedNotificationSchema = NotificationSchema.extend({ method: z.literal("notifications/initialized"), }); +export const isInitializedNotification = (value: unknown): value is InitializedNotification => + InitializedNotificationSchema.safeParse(value).success; + /* Ping */ /** * A ping, issued by either the server or the client, to check that the other party is still alive. The receiver must promptly respond, or else may be disconnected.