diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 1f5f59b6..44ee709b 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -158,6 +158,10 @@ importers: version: 2.9.0 workers/mcp-services: + dependencies: + '@modelcontextprotocol/server': + specifier: ^2.0.0 + version: 2.0.0 devDependencies: '@types/node': specifier: ^25.9.1 @@ -1110,6 +1114,10 @@ packages: '@mermaid-js/parser@1.1.1': resolution: {integrity: sha512-VuHdsYMK1bT6X2JbcAaWAhugTRvRBRyuZgd+c22swUeI9g/ntaxF7CY7dYarhZovofCbUNO0G7JesfmNtjYOCw==} + '@modelcontextprotocol/core@2.0.0': + resolution: {integrity: sha512-pJCEwGG7Lfr/+PQp9ZTwKXNeO5wzbfKL7H3MYpCorM4oFBoQrdjnBgEoqG+RjhsvS1FKrDbKux+M1HhlnGWqcA==} + engines: {node: '>=20'} + '@modelcontextprotocol/sdk@1.29.0': resolution: {integrity: sha512-zo37mZA9hJWpULgkRpowewez1y6ML5GsXJPY8FI0tBBCd77HEvza4jDqRKOXgHNn867PVGCyTdzqpz0izu5ZjQ==} engines: {node: '>=18'} @@ -1120,6 +1128,10 @@ packages: '@cfworker/json-schema': optional: true + '@modelcontextprotocol/server@2.0.0': + resolution: {integrity: sha512-YhHWdHfpFMQfd0prsEnxKeS3Qz3ytIGmsS0sth4KDjnacIT7hxk6hXHkJ9KysxlkvTM+WZAtQbbcUhdoP4Hvtw==} + engines: {node: '>=20'} + '@modelcontextprotocol/server@2.0.0-alpha.4': resolution: {integrity: sha512-/KEo3ZJ50HlagHp0lz2vPgfBZFtXHu6zTBXT9XqPc+4O9i4+IbBVWKq/a9yAfv/ifp0u3+mRo8SUk5kMKzhN8A==} engines: {node: '>=20'} @@ -6323,6 +6335,10 @@ snapshots: dependencies: '@chevrotain/types': 11.1.2 + '@modelcontextprotocol/core@2.0.0': + dependencies: + zod: 4.4.3 + '@modelcontextprotocol/sdk@1.29.0(@cfworker/json-schema@4.1.1)(zod@4.4.3)': dependencies: '@hono/node-server': 1.19.14(hono@4.12.27) @@ -6347,6 +6363,11 @@ snapshots: transitivePeerDependencies: - supports-color + '@modelcontextprotocol/server@2.0.0': + dependencies: + '@modelcontextprotocol/core': 2.0.0 + zod: 4.4.3 + '@modelcontextprotocol/server@2.0.0-alpha.4': dependencies: zod: 4.4.3 diff --git a/workers/mcp-services/README.md b/workers/mcp-services/README.md index ea9bf47b..2cb8f590 100644 --- a/workers/mcp-services/README.md +++ b/workers/mcp-services/README.md @@ -1,6 +1,8 @@ # mpp-services-mcp Read-only Cloudflare Worker MCP server for the MPP service discovery catalog. +The official `@modelcontextprotocol/server` SDK handles protocol validation and +Streamable HTTP transport; this package owns the discovery tools and catalog. Production endpoint: diff --git a/workers/mcp-services/package.json b/workers/mcp-services/package.json index a39d79fe..f950f404 100644 --- a/workers/mcp-services/package.json +++ b/workers/mcp-services/package.json @@ -12,6 +12,9 @@ "test": "vitest --run", "check": "pnpm gen:types && pnpm check:types && pnpm test" }, + "dependencies": { + "@modelcontextprotocol/server": "^2.0.0" + }, "devDependencies": { "@types/node": "^25.9.1", "typescript": "^6.0.3", diff --git a/workers/mcp-services/src/health.ts b/workers/mcp-services/src/health.ts index 231b021e..a00ab8d1 100644 --- a/workers/mcp-services/src/health.ts +++ b/workers/mcp-services/src/health.ts @@ -1,3 +1,4 @@ +import { LATEST_PROTOCOL_VERSION } from "@modelcontextprotocol/server"; import { workerMetrics } from "../../../src/lib/worker-metrics.js"; import type { WorkerEnv } from "./types.js"; @@ -121,7 +122,11 @@ async function assertHead(endpoint: string): Promise { } async function assertInitialize(endpoint: string): Promise { - const result = await rpc(endpoint, "initialize"); + const result = await rpc(endpoint, "initialize", { + protocolVersion: LATEST_PROTOCOL_VERSION, + capabilities: {}, + clientInfo: { name: "mpp-discovery-health", version: "1.0.0" }, + }); if (stringValue(object(result.serverInfo).name) !== "mpp-services-mcp") { throw new Error("initialize serverInfo mismatch"); } @@ -205,7 +210,7 @@ async function rpc( const body = await fetchJson(endpoint, { method: "POST", headers: { - accept: "application/json", + accept: "application/json, text/event-stream", "content-type": "application/json", }, body: JSON.stringify({ jsonrpc: "2.0", id: 1, method, params }), @@ -226,7 +231,19 @@ async function fetchJson( if (response.status !== 200) { throw new Error(`expected 200, received ${response.status}`); } - return object(await response.json()); + return object(await jsonRpcBody(response)); +} + +async function jsonRpcBody(response: Response): Promise { + if (response.headers.get("content-type")?.includes("application/json")) { + return response.json(); + } + const data = (await response.text()) + .split("\n") + .filter((line) => line.startsWith("data:")) + .at(-1); + if (!data) throw new Error("MCP response contained no JSON-RPC message"); + return JSON.parse(data.slice("data:".length)); } async function fetchWithTimeout( diff --git a/workers/mcp-services/src/index.test.ts b/workers/mcp-services/src/index.test.ts index 55d4bb65..2a228831 100644 --- a/workers/mcp-services/src/index.test.ts +++ b/workers/mcp-services/src/index.test.ts @@ -63,22 +63,13 @@ describe("worker routes", () => { it("handles MCP JSON-RPC at /mcp/services", async () => { const response = await worker.fetch( - new Request("https://worker.example.com/mcp/services", { - method: "POST", - headers: { "content-type": "application/json" }, - body: JSON.stringify({ - jsonrpc: "2.0", - id: 1, - method: "initialize", - params: {}, - }), - }), + initializeRequest(), envWithCatalog(), testContext(), ); expect(response.status).toBe(200); - const body = (await response.json()) as { + const body = (await jsonRpcBody(response)) as { result: { serverInfo: { name: string }; instructions: string }; }; expect(body.result.serverInfo.name).toBe("mpp-services-mcp"); @@ -99,16 +90,7 @@ describe("worker routes", () => { const log = vi.spyOn(console, "log").mockImplementation(() => {}); const response = await worker.fetch( - new Request("https://worker.example.com/mcp/services", { - method: "POST", - headers: { "content-type": "application/json" }, - body: JSON.stringify({ - jsonrpc: "2.0", - id: 1, - method: "initialize", - params: {}, - }), - }), + initializeRequest(), envWithCatalog(), testContext(), ); @@ -194,6 +176,38 @@ function envWithCatalog(): WorkerEnv { } as WorkerEnv; } +function initializeRequest(): Request { + return new Request("https://worker.example.com/mcp/services", { + method: "POST", + headers: { + accept: "application/json, text/event-stream", + "content-type": "application/json", + }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { + protocolVersion: "2025-06-18", + capabilities: {}, + clientInfo: { name: "mpp-worker-test", version: "1.0.0" }, + }, + }), + }); +} + +async function jsonRpcBody(response: Response): Promise { + if (response.headers.get("content-type")?.includes("application/json")) { + return response.json(); + } + const data = (await response.text()) + .split("\n") + .filter((line) => line.startsWith("data:")) + .at(-1); + if (!data) throw new Error("MCP response contained no JSON-RPC message"); + return JSON.parse(data.slice("data:".length)); +} + function testContext(): ExecutionContext { return { waitUntil() {}, diff --git a/workers/mcp-services/src/mcp.test.ts b/workers/mcp-services/src/mcp.test.ts index 0ad34fcb..b11ac461 100644 --- a/workers/mcp-services/src/mcp.test.ts +++ b/workers/mcp-services/src/mcp.test.ts @@ -101,6 +101,7 @@ describe("mcp handler", () => { expect(tool.outputSchema).toEqual( expect.objectContaining({ type: "object" }), ); + expect(tool.execution).toEqual({ taskSupport: "forbidden" }); } expect( tools.find((tool) => tool.name === "get_openapi")?.inputSchema, @@ -113,6 +114,98 @@ describe("mcp handler", () => { ); }); + it("serves 2026-era requests through the SDK handler", async () => { + const response = await handleMcp( + new Request("https://example.com/mcp", { + method: "POST", + headers: { + accept: "application/json", + "content-type": "application/json", + "mcp-method": "tools/list", + "mcp-protocol-version": "2026-07-28", + }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: 1, + method: "tools/list", + params: { + _meta: { + "io.modelcontextprotocol/protocolVersion": "2026-07-28", + "io.modelcontextprotocol/clientInfo": { + name: "mpp-worker-test", + version: "1.0.0", + }, + "io.modelcontextprotocol/clientCapabilities": {}, + }, + }, + }), + }), + envWithCatalog(), + testContext(), + ); + + expect(response.status).toBe(200); + expect(response.headers.get("content-type")).toContain("application/json"); + expect(response.headers.get("access-control-allow-origin")).toBe("*"); + const body = (await response.json()) as { + result: { tools: Array<{ name: string }> }; + }; + expect(body.result.tools).toHaveLength(11); + }); + + it("delegates HTTP and JSON-RPC validation to the SDK", async () => { + const missingAccept = await handleMcp( + new Request("https://example.com/mcp", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: 1, + method: "tools/list", + params: {}, + }), + }), + envWithCatalog(), + testContext(), + ); + expect(missingAccept.status).toBe(406); + expect(await missingAccept.json()).toEqual( + expect.objectContaining({ + error: expect.objectContaining({ code: -32000 }), + }), + ); + + const malformedJson = await handleMcp( + new Request("https://example.com/mcp", { + method: "POST", + headers: { + accept: "application/json, text/event-stream", + "content-type": "application/json", + }, + body: "{", + }), + envWithCatalog(), + testContext(), + ); + expect(malformedJson.status).toBe(400); + expect(await malformedJson.json()).toEqual( + expect.objectContaining({ + error: expect.objectContaining({ code: -32700 }), + }), + ); + + const notification = await handleMcp( + legacyRequest({ + jsonrpc: "2.0", + method: "notifications/initialized", + }), + envWithCatalog(), + testContext(), + ); + expect(notification.status).toBe(202); + expect(await notification.text()).toBe(""); + }); + it("searches endpoint-level payment offers with matching and ranking metadata", async () => { const body = await callTool("search_offers", { query: "inbox", @@ -332,12 +425,7 @@ describe("mcp handler", () => { ]) { const body = await callTool("search_services", args); expect(body.result.isError).toBe(true); - expect(body.result.structuredContent).toEqual( - expect.objectContaining({ - success: false, - error: expect.stringContaining("Allowed values:"), - }), - ); + expect(body.result.content[0]?.text).toContain("Input validation error"); } }); @@ -653,20 +741,17 @@ async function callTool(name: string, args: Record) { async function mcp(method: string, params: Record, env: Env) { const response = await handleMcp( - new Request("https://example.com/mcp", { - method: "POST", - headers: { "content-type": "application/json" }, - body: JSON.stringify({ jsonrpc: "2.0", id: 1, method, params }), - }), + legacyRequest({ jsonrpc: "2.0", id: 1, method, params }), env, testContext(), ); - return response.json() as Promise<{ + return jsonRpcBody(response) as Promise<{ result: { tools?: Array<{ name?: string; inputSchema?: unknown; outputSchema?: unknown; + execution?: unknown; }>; content: Array<{ type: string; text: string }>; isError?: boolean; @@ -688,6 +773,29 @@ async function mcp(method: string, params: Record, env: Env) { }>; } +function legacyRequest(body: unknown): Request { + return new Request("https://example.com/mcp", { + method: "POST", + headers: { + accept: "application/json, text/event-stream", + "content-type": "application/json", + }, + body: JSON.stringify(body), + }); +} + +async function jsonRpcBody(response: Response): Promise { + if (response.headers.get("content-type")?.includes("application/json")) { + return response.json(); + } + const data = (await response.text()) + .split("\n") + .filter((line) => line.startsWith("data:")) + .at(-1); + if (!data) throw new Error("MCP response contained no JSON-RPC message"); + return JSON.parse(data.slice("data:".length)); +} + function envWithCatalog(): Env { return envWithCatalogFor(services); } diff --git a/workers/mcp-services/src/mcp.ts b/workers/mcp-services/src/mcp.ts index ad7f6e18..9d54af84 100644 --- a/workers/mcp-services/src/mcp.ts +++ b/workers/mcp-services/src/mcp.ts @@ -1,3 +1,13 @@ +import { + type CallToolResult, + createMcpHandler, + fromJsonSchema, + LATEST_PROTOCOL_VERSION, + McpServer, + ProtocolError, + ProtocolErrorCode, + type ToolExecution, +} from "@modelcontextprotocol/server"; import { workerMetrics } from "../../../src/lib/worker-metrics.js"; import { getCatalog } from "./cache.js"; import { @@ -22,7 +32,7 @@ import { type WorkerEnv, } from "./types.js"; -const PROTOCOL_VERSION = "2025-06-18"; +const PROTOCOL_VERSION = LATEST_PROTOCOL_VERSION; const SERVER_VERSION = "1.0.0"; const ADVISORY = "Discovery is advisory; the runtime 402 Challenge is authoritative."; @@ -101,24 +111,19 @@ const INITIALIZE_INSTRUCTIONS = [ "This server does not register services, execute payments, authorize requests, or replace runtime 402 Challenge validation.", ].join(" "); -type JsonRpcId = string | number | null; - -type JsonRpcRequest = { - jsonrpc?: string; - id?: JsonRpcId; - method?: string; - params?: unknown; +type JsonRpcMetricContext = { + method?: unknown; + toolName?: unknown; }; -type ToolCallParams = { - name?: unknown; - arguments?: unknown; +type DiscoveryTool = { + name: string; + description: string; + inputSchema: Record; + outputSchema: Record; + execution?: ToolExecution; }; -type JsonRpcResponsePayload = - | { jsonrpc: "2.0"; id: JsonRpcId; result: unknown } - | { jsonrpc: "2.0"; id: JsonRpcId; error: { code: number; message: string } }; - type Pagination = { limit: number; offset: number; @@ -160,27 +165,66 @@ export async function handleMcp( env: WorkerEnv, ctx: ExecutionContext, ): Promise { - let payload: unknown; - try { - payload = await request.json(); - } catch { - recordJsonRpcError(undefined, "-32700"); - return jsonResponse(jsonRpcError(null, -32700, "Parse error")); + const startedAt = Date.now(); + const metricContext = await jsonRpcMetricContext(request); + let toolDispatched = false; + const handler = createMcpHandler(() => { + const server = mcpServer(env, ctx, () => { + toolDispatched = true; + }); + server.server.fallbackRequestHandler = (message) => { + recordJsonRpcError(message.method, "-32601"); + throw new ProtocolError( + ProtocolErrorCode.MethodNotFound, + `Method not found: ${message.method}`, + ); + }; + return server; + }); + + const response = await handler.fetch(request); + if (metricContext.method === "tools/call" && !toolDispatched) { + recordJsonRpcError("tools/call", "tool_error", metricContext.toolName); + recordToolCallMetrics( + metricToolNameFor(metricContext.toolName), + "error", + Date.now() - startedAt, + ); } + return withCors(response); +} - if (Array.isArray(payload)) { - const responses: JsonRpcResponsePayload[] = []; - for (const item of payload) { - const response = await handleMessage(asRequest(item), env, ctx); - if (response) responses.push(response); - } - if (responses.length === 0) return emptyAcceptedResponse(); - return jsonResponse(responses); +function mcpServer( + env: WorkerEnv, + ctx: ExecutionContext, + onToolDispatch: () => void, +): McpServer { + const server = new McpServer(serverInfo(), { + capabilities: { tools: { listChanged: false } }, + instructions: INITIALIZE_INSTRUCTIONS, + }); + + for (const tool of toolSchemas()) { + const registered = server.registerTool( + tool.name, + { + description: tool.description, + inputSchema: sdkSchema(tool.inputSchema), + outputSchema: sdkSchema(tool.outputSchema), + }, + async (args) => { + onToolDispatch(); + return handleToolCall(tool.name, args, env, ctx); + }, + ); + registered.execution = tool.execution; } - const response = await handleMessage(asRequest(payload), env, ctx); - if (!response) return emptyAcceptedResponse(); - return jsonResponse(response); + return server; +} + +function sdkSchema(schema: Record) { + return fromJsonSchema(schema as Parameters[0]); } export function serverCard(endpoint: string) { @@ -212,15 +256,25 @@ export function serverCard(endpoint: string) { } export function jsonHeaders(extra?: HeadersInit): Headers { - const headers = new Headers(extra); + const headers = corsHeaders(new Headers(extra)); headers.set("content-type", "application/json"); + return headers; +} + +function corsHeaders(headers: Headers): Headers { headers.set("access-control-allow-origin", "*"); headers.set("access-control-allow-methods", "GET,HEAD,POST,OPTIONS"); headers.set( "access-control-allow-headers", - "content-type,mcp-protocol-version", + "accept,content-type,mcp-method,mcp-name,mcp-protocol-version", ); - headers.set("mcp-protocol-version", PROTOCOL_VERSION); + headers.set( + "access-control-expose-headers", + "mcp-protocol-version,mcp-session-id", + ); + if (!headers.has("mcp-protocol-version")) { + headers.set("mcp-protocol-version", PROTOCOL_VERSION); + } return headers; } @@ -228,53 +282,43 @@ export function optionsResponse(): Response { return new Response(null, { status: 204, headers: jsonHeaders() }); } -async function handleMessage( - request: JsonRpcRequest | undefined, - env: WorkerEnv, - ctx: ExecutionContext, -): Promise { - if (request?.jsonrpc !== "2.0" || typeof request.method !== "string") { - recordJsonRpcError(request?.method, "-32600"); - return jsonRpcError(request?.id ?? null, -32600, "Invalid Request"); +async function jsonRpcMetricContext( + request: Request, +): Promise { + if (request.method !== "POST") return {}; + try { + const body = await request.clone().json(); + if (!isRecord(body)) return {}; + const params = isRecord(body.params) ? body.params : {}; + if (body.jsonrpc !== "2.0" || typeof body.method !== "string") { + recordJsonRpcError(body.method, "-32600"); + } + return { + method: body.method, + toolName: params.name, + }; + } catch { + recordJsonRpcError(undefined, "-32700"); + return {}; } +} - switch (request.method) { - case "initialize": - return jsonRpcResult(request.id ?? null, initializeResult()); - case "notifications/initialized": - return undefined; - case "ping": - return jsonRpcResult(request.id ?? null, {}); - case "tools/list": - return jsonRpcResult(request.id ?? null, { tools: toolSchemas() }); - case "tools/call": - return jsonRpcResult( - request.id ?? null, - await handleToolCall(toolCallParams(request.params), env, ctx), - ); - case "resources/list": - return jsonRpcResult(request.id ?? null, { resources: [] }); - case "resources/templates/list": - return jsonRpcResult(request.id ?? null, { resourceTemplates: [] }); - case "prompts/list": - return jsonRpcResult(request.id ?? null, { prompts: [] }); - default: - recordJsonRpcError(request.method, "-32601"); - return jsonRpcError( - request.id ?? null, - -32601, - `Method not found: ${request.method}`, - ); - } +function withCors(response: Response): Response { + const headers = corsHeaders(new Headers(response.headers)); + return new Response(response.body, { + status: response.status, + statusText: response.statusText, + headers, + }); } async function handleToolCall( - params: ToolCallParams, + name: string, + input: unknown, env: WorkerEnv, ctx: ExecutionContext, -) { - const name = typeof params.name === "string" ? params.name : ""; - const args = objectArgs(params.arguments); +): Promise { + const args = objectArgs(input); const metricToolName = metricToolNameFor(name); const startedAt = Date.now(); let outcome = "success"; @@ -767,20 +811,6 @@ function isRecord(value: unknown): value is JsonObject { return typeof value === "object" && value !== null && !Array.isArray(value); } -function initializeResult() { - // This server supports exactly one protocol version, so the response always - // advertises PROTOCOL_VERSION. Per the MCP spec this is the correct reply - // whether or not the client requested that same version. - return { - protocolVersion: PROTOCOL_VERSION, - capabilities: { - tools: {}, - }, - serverInfo: serverInfo(), - instructions: INITIALIZE_INSTRUCTIONS, - }; -} - function serverInfo() { return { name: "mpp-services-mcp", @@ -1064,7 +1094,7 @@ function equalsIgnoreCase(left: string, right: string): boolean { return normalizedText(left) === normalizedText(right); } -function toolSchemas() { +function toolSchemas(): DiscoveryTool[] { const advisory = ` ${ADVISORY}`; return [ { @@ -2000,14 +2030,17 @@ function oneOfSuccessOrError(successSchema: Record) { }; } -function toolResult(structuredContent: unknown, text: string) { +function toolResult( + structuredContent: Record, + text: string, +): CallToolResult { return { content: [{ type: "text", text }], structuredContent, }; } -function toolError(message: string) { +function toolError(message: string): CallToolResult { return { content: [{ type: "text", text: `${message}. ${ADVISORY}` }], structuredContent: { success: false, error: message, advisory: ADVISORY }, @@ -2225,20 +2258,6 @@ function objectArgs(value: unknown): Record { return {}; } -function toolCallParams(value: unknown): ToolCallParams { - if (typeof value === "object" && value !== null && !Array.isArray(value)) { - return value as ToolCallParams; - } - return {}; -} - -function asRequest(value: unknown): JsonRpcRequest | undefined { - if (typeof value !== "object" || value === null || Array.isArray(value)) { - return undefined; - } - return value as JsonRpcRequest; -} - function recordJsonRpcError( method: unknown, errorCode: string, @@ -2275,26 +2294,6 @@ function metricToolNameFor(name: unknown): string { return METRIC_TOOL_NAMES.has(name) ? name : "unknown"; } -function jsonRpcResult(id: JsonRpcId, result: unknown): JsonRpcResponsePayload { - return { jsonrpc: "2.0", id, result }; -} - -function jsonRpcError( - id: JsonRpcId, - code: number, - message: string, -): JsonRpcResponsePayload { - return { jsonrpc: "2.0", id, error: { code, message } }; -} - -function jsonResponse(payload: unknown): Response { - return new Response(JSON.stringify(payload), { headers: jsonHeaders() }); -} - -function emptyAcceptedResponse(): Response { - return new Response(null, { status: 202, headers: jsonHeaders() }); -} - function errorMessage(error: unknown): string { return error instanceof Error ? error.message : String(error); }