mirror of
https://github.com/morluto/rea.git
synced 2026-11-03 05:16:16 +00:00
fix(mcp): match Accept ranges exactly and keep a session whose shutdown failed
Addresses the fourth automated review round: a sessionless POST replaces a session only when its Accept header lists application/json and text/event-stream as exact media ranges, and a session whose transport shutdown rejected stays owned so the next close retries it. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
1 parent
b76f354d76
commit
fb22df370b
3 files changed
+73
-13
No files matched your search
@@ -216,3 +216,11 @@ allowlist (test: `ip6-localhost` bound to `::1`, `dev.local` bound to
|
||||
`127.0.0.1`); and the MCP runtime passes `closeInstance: false` so the factory
|
||||
owner in `startMcpTransport` remains the single owner of instance cleanup
|
||||
(test: `close` leaves the instance open until its owner closes it).
|
||||
|
||||
A fourth round (Codex, two findings) was addressed next: the replacement
|
||||
guard parses `Accept` into media ranges and requires exact `application/json`
|
||||
and `text/event-stream` entries (test: `application/json-patch+json,
|
||||
text/event-streaming` is refused while the connected client keeps echoing),
|
||||
and a session whose transport shutdown rejected stays owned so the next close
|
||||
retries it (test: a failing transport close rejects `close()` once, the second
|
||||
call completes, the shutdown hook fires once and later requests get `503`).
|
||||
@@ -2,9 +2,12 @@ import {
|
||||
Client,
|
||||
StreamableHTTPClientTransport,
|
||||
} from "@modelcontextprotocol/client";
|
||||
import { McpServer } from "@modelcontextprotocol/server";
|
||||
import {
|
||||
McpServer,
|
||||
WebStandardStreamableHTTPServerTransport,
|
||||
} from "@modelcontextprotocol/server";
|
||||
import pino from "pino";
|
||||
import { afterEach, expect, it } from "vitest";
|
||||
import { afterEach, expect, it, vi } from "vitest";
|
||||
import { z } from "zod";
|
||||
|
||||
import {
|
||||
@@ -421,3 +424,48 @@ it("leaves the instance to its owner when closeInstance is false", async () => {
|
||||
await server.close();
|
||||
expect(instanceCloses).toBe(1);
|
||||
});
|
||||
|
||||
it("requires exact Accept media ranges before a sessionless POST may replace the session", async () => {
|
||||
const { served } = await start();
|
||||
const { client } = await connect(served.url);
|
||||
const url = new URL(served.url);
|
||||
const lookalike = await rawHttpRequest(url, {
|
||||
method: "POST",
|
||||
headers: {
|
||||
host: `127.0.0.1:${url.port}`,
|
||||
"content-type": "application/json",
|
||||
accept: "application/json-patch+json, text/event-streaming",
|
||||
},
|
||||
body: await initializeRequest().text(),
|
||||
});
|
||||
expect(lookalike.status).toBeGreaterThanOrEqual(400);
|
||||
expect(lookalike.status).toBeLessThan(500);
|
||||
expect(await echo(client, "still mine")).toEqual([
|
||||
{ type: "text", text: "still mine" },
|
||||
]);
|
||||
await client.close();
|
||||
});
|
||||
|
||||
it("keeps a session whose transport shutdown failed so the next close retries it", async () => {
|
||||
let shutdowns = 0;
|
||||
const slot = new McpSessionSlot(
|
||||
createEchoServer(() => {
|
||||
shutdowns += 1;
|
||||
}),
|
||||
undefined,
|
||||
);
|
||||
// The slot binds the transport's close when it arms, so the spy must exist before the first request.
|
||||
const close = vi
|
||||
.spyOn(WebStandardStreamableHTTPServerTransport.prototype, "close")
|
||||
.mockImplementationOnce(() => Promise.reject(new Error("transport stuck")));
|
||||
try {
|
||||
await slot.handle(initializeRequest());
|
||||
await expect(slot.close()).rejects.toThrow("transport stuck");
|
||||
expect(shutdowns).toBe(0);
|
||||
await expect(slot.close()).resolves.toBeUndefined();
|
||||
expect(shutdowns).toBe(1);
|
||||
expect((await slot.handle(initializeRequest())).status).toBe(503);
|
||||
} finally {
|
||||
close.mockRestore();
|
||||
}
|
||||
});
|
||||
@@ -108,15 +108,18 @@ const SHUTTING_DOWN = (): Response =>
|
||||
jsonError(503, -32000, "Server shutting down");
|
||||
|
||||
/** The headers the SDK requires of a streamable POST; a request without them is left to the transport's refusal. */
|
||||
/** The media type proper of one header value or Accept range: no parameters, lowercase. */
|
||||
const mediaType = (value: string): string =>
|
||||
(value.split(";")[0] ?? "").trim().toLowerCase();
|
||||
const hasStreamablePostHeaders = (request: Request): boolean => {
|
||||
const accept = request.headers.get("accept") ?? "";
|
||||
const contentType = request.headers.get("content-type") ?? "";
|
||||
// The media type proper, without parameters: `application/json-patch+json` is not JSON here.
|
||||
const essence = (contentType.split(";")[0] ?? "").trim().toLowerCase();
|
||||
// Exact media ranges and type: `application/json-patch+json` or `text/event-streaming` do not count.
|
||||
const accepted = new Set(
|
||||
(request.headers.get("accept") ?? "").split(",").map(mediaType),
|
||||
);
|
||||
return (
|
||||
accept.includes("application/json") &&
|
||||
accept.includes("text/event-stream") &&
|
||||
essence === "application/json"
|
||||
accepted.has("application/json") &&
|
||||
accepted.has("text/event-stream") &&
|
||||
mediaType(request.headers.get("content-type") ?? "") === "application/json"
|
||||
);
|
||||
};
|
||||
|
||||
@@ -187,8 +190,9 @@ export class McpSessionSlot {
|
||||
this.#closed = true;
|
||||
await this.#arming?.catch(() => undefined);
|
||||
const open = this.#live;
|
||||
this.#live = undefined;
|
||||
// The session stays owned until its shutdown succeeds, so a retry can finish it.
|
||||
await open?.shutdown();
|
||||
if (this.#live === open) this.#live = undefined;
|
||||
}
|
||||
|
||||
#current(): Promise<WebStandardStreamableHTTPServerTransport> {
|
||||
@@ -225,9 +229,10 @@ export class McpSessionSlot {
|
||||
protocol.onclose = NO_SHUTDOWN_HOOK;
|
||||
try {
|
||||
await shutdown();
|
||||
// Only a completed shutdown releases the session; a failed one stays live for a retry.
|
||||
if (this.#live?.transport === transport) this.#live = undefined;
|
||||
} finally {
|
||||
protocol.onclose = this.#shutdownHook;
|
||||
if (this.#live?.transport === transport) this.#live = undefined;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -308,8 +313,7 @@ export const serveStreamableHttp = async (
|
||||
const bound = httpServer.address();
|
||||
let url: string;
|
||||
try {
|
||||
url = endpointUrl(bound);
|
||||
origin = new URL(url).origin;
|
||||
((url = endpointUrl(bound)), (origin = new URL(url).origin));
|
||||
} catch (cause: unknown) {
|
||||
await closeListener().catch(() => undefined);
|
||||
throw new StreamableHttpListenError(cause);
|
||||
|
||||
Reference in new issue
Block a user