fix: address Copilot review round 2 on #2318
- fetchTracking: media types and HTTP methods compare case-insensitively; a `Request` input's own method is honoured; the SSE line scanner accepts LF, CRLF and a bare CR, holding a trailing CR until the next chunk decides whether it is half of a CRLF. - MessageTrackingTransport does not report a send failure for a request the caller aborted through `requestSignal`, and InspectorClient keeps a request whose send failed with a request timeout — both are sent, unanswered requests, which is the state to keep reporting. - createTransportNode builds a fetch tracker when any tracking callback is supplied, not only `onFetchRequest`. - RemoteClientTransport's relay wait expires with the SDK's own timeout shape, so the web client's relay timeout is annotated like every other. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01APKSYHGxXBUpnR7ErFiLJn Signed-off-by: cliffhall <cliff@futurescale.com>
This commit is contained in:
parent
1a5dd6820d
commit
c565b56fcd
11 changed files with 266 additions and 27 deletions
|
|
@ -117,7 +117,7 @@ describe("RemoteClientTransport send contract", () => {
|
||||||
const started = Date.now();
|
const started = Date.now();
|
||||||
await expect(
|
await expect(
|
||||||
transport.send({ jsonrpc: "2.0", id: 1, method: "tools/list" }),
|
transport.send({ jsonrpc: "2.0", id: 1, method: "tools/list" }),
|
||||||
).rejects.toThrow(/Timed out waiting for MCP response on SSE/i);
|
).rejects.toThrow(/Request timed out/);
|
||||||
expect(Date.now() - started).toBeLessThan(5000);
|
expect(Date.now() - started).toBeLessThan(5000);
|
||||||
|
|
||||||
await transport.close();
|
await transport.close();
|
||||||
|
|
|
||||||
|
|
@ -826,6 +826,56 @@ describe("createFetchTracker long-lived stream watching (#2318)", () => {
|
||||||
expect(last.closedAt).toBeInstanceOf(Date);
|
expect(last.closedAt).toBeInstanceOf(Date);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("accepts bare CR line endings, and a CRLF split across chunks", async () => {
|
||||||
|
const server = controlledStream();
|
||||||
|
const { updates } = await trackStream(server.stream);
|
||||||
|
server.push("data: a\r\rdata: b\r");
|
||||||
|
await flush();
|
||||||
|
// The trailing CR is held: it may be the first half of a CRLF.
|
||||||
|
expect(updates.map((u) => u.stream.eventCount)).toEqual([0, 1]);
|
||||||
|
server.push("\n\r\n");
|
||||||
|
await flush();
|
||||||
|
expect(updates.map((u) => u.stream.eventCount)).toEqual([0, 1, 2]);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("classifies the content type and the method case-insensitively", async () => {
|
||||||
|
const server = controlledStream();
|
||||||
|
const updates: unknown[] = [];
|
||||||
|
const fetcher = createFetchTracker(
|
||||||
|
(async () =>
|
||||||
|
new Response(server.stream, {
|
||||||
|
headers: { "content-type": "Text/Event-Stream" },
|
||||||
|
})) as typeof fetch,
|
||||||
|
{ updateStream: (id, state) => updates.push({ id, state }) },
|
||||||
|
);
|
||||||
|
await fetcher("https://example.com/mcp", { method: "get" });
|
||||||
|
expect(updates).toHaveLength(1);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("takes the method from a Request input when init carries none", async () => {
|
||||||
|
// A bounded POST reply must go down the body-capture path, not the
|
||||||
|
// watcher, when the caller passed a `Request` rather than `init`.
|
||||||
|
const bodies: string[] = [];
|
||||||
|
const updates: unknown[] = [];
|
||||||
|
const tracked: FetchRequestEntryBase[] = [];
|
||||||
|
const fetcher = createFetchTracker(
|
||||||
|
(async () =>
|
||||||
|
new Response("data: x\n\n", {
|
||||||
|
headers: { "content-type": "text/event-stream" },
|
||||||
|
})) as typeof fetch,
|
||||||
|
{
|
||||||
|
trackRequest: (entry) => tracked.push(entry),
|
||||||
|
updateResponseBody: (_id, body) => bodies.push(body),
|
||||||
|
updateStream: (id, state) => updates.push({ id, state }),
|
||||||
|
},
|
||||||
|
);
|
||||||
|
await fetcher(new Request("https://example.com/mcp", { method: "POST" }));
|
||||||
|
await flush();
|
||||||
|
expect(tracked[0]?.method).toBe("POST");
|
||||||
|
expect(bodies).toEqual(["data: x\n\n"]);
|
||||||
|
expect(updates).toEqual([]);
|
||||||
|
});
|
||||||
|
|
||||||
it("reports the close when the server ends the stream", async () => {
|
it("reports the close when the server ends the stream", async () => {
|
||||||
const server = controlledStream();
|
const server = controlledStream();
|
||||||
const { updates, id } = await trackStream(server.stream);
|
const { updates, id } = await trackStream(server.stream);
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,75 @@
|
||||||
|
import { describe, it, expect } from "vitest";
|
||||||
|
import { SdkError, SdkErrorCode } from "@modelcontextprotocol/client";
|
||||||
|
import type { JSONRPCRequest } from "@modelcontextprotocol/client";
|
||||||
|
import { InspectorClient } from "@inspector/core/mcp/inspectorClient.js";
|
||||||
|
import type { MessageTrackingCallbacks } from "@inspector/core/mcp/messageTrackingTransport.js";
|
||||||
|
|
||||||
|
/**
|
||||||
|
* The outstanding-request bookkeeping behind the connection diagnostics
|
||||||
|
* (#2318), driven through the message-tracking callbacks the client hands its
|
||||||
|
* transport — the same seam the transport uses, without a connection.
|
||||||
|
*/
|
||||||
|
describe("InspectorClient diagnostics tracking callbacks", () => {
|
||||||
|
interface Internals {
|
||||||
|
createMessageTrackingCallbacks: () => MessageTrackingCallbacks;
|
||||||
|
}
|
||||||
|
|
||||||
|
function makeClient(): {
|
||||||
|
client: InspectorClient;
|
||||||
|
callbacks: MessageTrackingCallbacks;
|
||||||
|
} {
|
||||||
|
const client = new InspectorClient(
|
||||||
|
{ type: "stdio", command: "noop", args: [] },
|
||||||
|
// Never connects; the callbacks are exercised directly.
|
||||||
|
{ environment: { transport: () => ({}) as never } },
|
||||||
|
);
|
||||||
|
const callbacks = (
|
||||||
|
client as unknown as Internals
|
||||||
|
).createMessageTrackingCallbacks();
|
||||||
|
return { client, callbacks };
|
||||||
|
}
|
||||||
|
|
||||||
|
const request = (id: number): JSONRPCRequest => ({
|
||||||
|
jsonrpc: "2.0",
|
||||||
|
id,
|
||||||
|
method: "tools/list",
|
||||||
|
});
|
||||||
|
|
||||||
|
const outstandingIds = (client: InspectorClient) =>
|
||||||
|
client.getConnectionDiagnostics().outstandingRequests.map((r) => r.id);
|
||||||
|
|
||||||
|
it("rolls a request back when its send failed before reaching the wire", () => {
|
||||||
|
const { client, callbacks } = makeClient();
|
||||||
|
callbacks.trackRequest!(request(1), "client");
|
||||||
|
expect(outstandingIds(client)).toEqual([1]);
|
||||||
|
callbacks.trackSendFailure!(request(1), new Error("connection closed"));
|
||||||
|
expect(outstandingIds(client)).toEqual([]);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("keeps a request whose send failed with a request timeout", () => {
|
||||||
|
// The browser's remote transport awaits the response inside `send`, so
|
||||||
|
// its relay timeout surfaces as a send failure — the request reached the
|
||||||
|
// server and is still unanswered.
|
||||||
|
const { client, callbacks } = makeClient();
|
||||||
|
callbacks.trackRequest!(request(1), "client");
|
||||||
|
callbacks.trackSendFailure!(
|
||||||
|
request(1),
|
||||||
|
new SdkError(SdkErrorCode.RequestTimeout, "Request timed out", {
|
||||||
|
timeout: 60_000,
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
expect(outstandingIds(client)).toEqual([1]);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("only tracks client-originated requests, and ignores a rollback for an unknown id", () => {
|
||||||
|
const { client, callbacks } = makeClient();
|
||||||
|
callbacks.trackRequest!(request(7), "server");
|
||||||
|
expect(outstandingIds(client)).toEqual([]);
|
||||||
|
let changes = 0;
|
||||||
|
client.addEventListener("connectionDiagnosticsChange", () => {
|
||||||
|
changes += 1;
|
||||||
|
});
|
||||||
|
callbacks.trackSendFailure!(request(7), new Error("nope"));
|
||||||
|
expect(changes).toBe(0);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
@ -48,6 +48,18 @@ describe("MessageTrackingTransport.send", () => {
|
||||||
await expect(failing.send(request)).rejects.toBe(boom);
|
await expect(failing.send(request)).rejects.toBe(boom);
|
||||||
expect(callbacks.trackRequest).toHaveBeenCalledWith(request, "client");
|
expect(callbacks.trackRequest).toHaveBeenCalledWith(request, "client");
|
||||||
expect(trackSendFailure).toHaveBeenCalledWith(request, boom);
|
expect(trackSendFailure).toHaveBeenCalledWith(request, boom);
|
||||||
|
// A caller's own abort is a cancellation, not a send failure: the
|
||||||
|
// request may well have reached the server, and stays unanswered.
|
||||||
|
trackSendFailure.mockClear();
|
||||||
|
const controller = new AbortController();
|
||||||
|
controller.abort();
|
||||||
|
await expect(
|
||||||
|
failing.send(
|
||||||
|
{ jsonrpc: "2.0", id: 2, method: "tools/call" },
|
||||||
|
{ requestSignal: controller.signal },
|
||||||
|
),
|
||||||
|
).rejects.toBe(boom);
|
||||||
|
expect(trackSendFailure).not.toHaveBeenCalled();
|
||||||
// Only requests roll back: a failed notification has no pending entry.
|
// Only requests roll back: a failed notification has no pending entry.
|
||||||
trackSendFailure.mockClear();
|
trackSendFailure.mockClear();
|
||||||
await expect(
|
await expect(
|
||||||
|
|
|
||||||
|
|
@ -188,6 +188,35 @@ describe("Transport", () => {
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("tracks fetches when only onFetchResponseBody is supplied (#2318)", async () => {
|
||||||
|
// Any one of the tracking callbacks earns the tracker — a caller that
|
||||||
|
// wants only bodies (or only stream updates) must not be handed the
|
||||||
|
// bare fetch and silently receive nothing.
|
||||||
|
const server = createTestServerHttp({
|
||||||
|
serverInfo: createTestServerInfo(),
|
||||||
|
tools: [createEchoTool()],
|
||||||
|
serverType: "streamable-http",
|
||||||
|
});
|
||||||
|
try {
|
||||||
|
await server.start();
|
||||||
|
const bodies: string[] = [];
|
||||||
|
const result = createTransportNode(
|
||||||
|
{ type: "streamable-http", url: server.url },
|
||||||
|
{ onFetchResponseBody: (_id, body) => bodies.push(body) },
|
||||||
|
);
|
||||||
|
const client = new Client(
|
||||||
|
{ name: "test-client", version: "1.0.0" },
|
||||||
|
{ capabilities: {} },
|
||||||
|
);
|
||||||
|
await client.connect(result.transport);
|
||||||
|
await client.listTools();
|
||||||
|
await client.close();
|
||||||
|
expect(bodies.some((body) => body.includes("tools"))).toBe(true);
|
||||||
|
} finally {
|
||||||
|
await server.stop();
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
it("should call onFetchRequest callback for streamable-http transport", async () => {
|
it("should call onFetchRequest callback for streamable-http transport", async () => {
|
||||||
const server = createTestServerHttp({
|
const server = createTestServerHttp({
|
||||||
serverInfo: createTestServerInfo(),
|
serverInfo: createTestServerInfo(),
|
||||||
|
|
|
||||||
|
|
@ -16,6 +16,7 @@ import type { RemoteTransportOptions } from "@inspector/core/mcp/remote/remoteCl
|
||||||
import type { MCPServerConfig } from "@inspector/core/mcp/types.js";
|
import type { MCPServerConfig } from "@inspector/core/mcp/types.js";
|
||||||
import type { RemoteEvent } from "@inspector/core/mcp/remote/types.js";
|
import type { RemoteEvent } from "@inspector/core/mcp/remote/types.js";
|
||||||
import type { JSONRPCMessage } from "@modelcontextprotocol/client";
|
import type { JSONRPCMessage } from "@modelcontextprotocol/client";
|
||||||
|
import { SdkErrorCode } from "@modelcontextprotocol/client";
|
||||||
import {
|
import {
|
||||||
AuthChallengeError,
|
AuthChallengeError,
|
||||||
AuthRecoveryRequiredError,
|
AuthRecoveryRequiredError,
|
||||||
|
|
@ -615,6 +616,27 @@ describe("RemoteClientTransport (focused branch coverage)", () => {
|
||||||
).rejects.toThrow(/Transport is closed/);
|
).rejects.toThrow(/Transport is closed/);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("rejects an expired SSE response wait with the SDK's own timeout shape (#2318)", async () => {
|
||||||
|
// The relay's budget, not the request's, can be the one that expires;
|
||||||
|
// shaped as the SDK's timeout so InspectorClient annotates it the same.
|
||||||
|
const t = makeTransport(
|
||||||
|
{
|
||||||
|
events: () => openEventsResponse(),
|
||||||
|
send: () => jsonResponse({ ok: true }),
|
||||||
|
},
|
||||||
|
{ sseResponseTimeoutMs: 50 },
|
||||||
|
);
|
||||||
|
await t.start();
|
||||||
|
await expect(
|
||||||
|
t.send({ jsonrpc: "2.0", id: 1, method: "tools/list" }),
|
||||||
|
).rejects.toMatchObject({
|
||||||
|
code: SdkErrorCode.RequestTimeout,
|
||||||
|
message: "Request timed out",
|
||||||
|
data: { timeout: 50 },
|
||||||
|
});
|
||||||
|
await t.close();
|
||||||
|
});
|
||||||
|
|
||||||
it("does not wait for an SSE response on subscriptions/listen (long-lived stream)", async () => {
|
it("does not wait for an SSE response on subscriptions/listen (long-lived stream)", async () => {
|
||||||
// A `subscriptions/listen` never yields a JSON-RPC response — only an
|
// A `subscriptions/listen` never yields a JSON-RPC response — only an
|
||||||
// acknowledged notification over the event channel — so send() must
|
// acknowledged notification over the event channel — so send() must
|
||||||
|
|
|
||||||
|
|
@ -222,11 +222,13 @@ export function isLongLivedStreamResponse(
|
||||||
method: string,
|
method: string,
|
||||||
contentType: string | null | undefined,
|
contentType: string | null | undefined,
|
||||||
): boolean {
|
): boolean {
|
||||||
if (method !== "GET") return false;
|
if (method.toUpperCase() !== "GET") return false;
|
||||||
if (!contentType) return false;
|
if (!contentType) return false;
|
||||||
|
// Media types are case-insensitive (RFC 9110 §8.3.1); a server may send
|
||||||
|
// `Text/Event-Stream` and still be sending an event stream.
|
||||||
|
const type = contentType.toLowerCase();
|
||||||
return (
|
return (
|
||||||
contentType.includes("text/event-stream") ||
|
type.includes("text/event-stream") || type.includes("application/x-ndjson")
|
||||||
contentType.includes("application/x-ndjson")
|
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -261,7 +263,9 @@ export type LongLivedStreamFraming = "sse" | "ndjson";
|
||||||
export function longLivedStreamFraming(
|
export function longLivedStreamFraming(
|
||||||
contentType: string | null | undefined,
|
contentType: string | null | undefined,
|
||||||
): LongLivedStreamFraming {
|
): LongLivedStreamFraming {
|
||||||
return contentType?.includes("application/x-ndjson") ? "ndjson" : "sse";
|
return contentType?.toLowerCase().includes("application/x-ndjson")
|
||||||
|
? "ndjson"
|
||||||
|
: "sse";
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
@ -314,8 +318,7 @@ function watchLongLivedStream(
|
||||||
// stream with no state to show — the count only moves on an event.
|
// stream with no state to show — the count only moves on an event.
|
||||||
update(id, { eventCount });
|
update(id, { eventCount });
|
||||||
|
|
||||||
const consumeLine = (rawLine: string): void => {
|
const consumeLine = (line: string): void => {
|
||||||
const line = rawLine.endsWith("\r") ? rawLine.slice(0, -1) : rawLine;
|
|
||||||
if (framing === "ndjson") {
|
if (framing === "ndjson") {
|
||||||
if (line.trim() === "") return;
|
if (line.trim() === "") return;
|
||||||
eventCount += 1;
|
eventCount += 1;
|
||||||
|
|
@ -335,18 +338,35 @@ function watchLongLivedStream(
|
||||||
if (line === "data" || line.startsWith("data:")) blockHasData = true;
|
if (line === "data" || line.startsWith("data:")) blockHasData = true;
|
||||||
};
|
};
|
||||||
|
|
||||||
|
// Split off every complete line in `pending`. The SSE grammar (and the
|
||||||
|
// SDK's parser) accept LF, CRLF and a bare CR as a line ending, so all three
|
||||||
|
// end a line here; a CR that is the last byte so far is held back, since
|
||||||
|
// the LF of a CRLF pair may still be in flight in the next chunk.
|
||||||
|
const consumeCompleteLines = (): void => {
|
||||||
|
for (;;) {
|
||||||
|
const cr = pending.indexOf("\r");
|
||||||
|
const lf = pending.indexOf("\n");
|
||||||
|
const end = cr === -1 ? lf : lf === -1 ? cr : Math.min(cr, lf);
|
||||||
|
if (end === -1) return;
|
||||||
|
if (pending[end] === "\r") {
|
||||||
|
if (end === pending.length - 1) return;
|
||||||
|
const width = pending[end + 1] === "\n" ? 2 : 1;
|
||||||
|
consumeLine(pending.slice(0, end));
|
||||||
|
pending = pending.slice(end + width);
|
||||||
|
} else {
|
||||||
|
consumeLine(pending.slice(0, end));
|
||||||
|
pending = pending.slice(end + 1);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
const pump = async (): Promise<void> => {
|
const pump = async (): Promise<void> => {
|
||||||
try {
|
try {
|
||||||
for (;;) {
|
for (;;) {
|
||||||
const { value, done } = await reader.read();
|
const { value, done } = await reader.read();
|
||||||
if (done) break;
|
if (done) break;
|
||||||
pending += decoder.decode(value, { stream: true });
|
pending += decoder.decode(value, { stream: true });
|
||||||
let newline = pending.indexOf("\n");
|
consumeCompleteLines();
|
||||||
while (newline !== -1) {
|
|
||||||
consumeLine(pending.slice(0, newline));
|
|
||||||
pending = pending.slice(newline + 1);
|
|
||||||
newline = pending.indexOf("\n");
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
} catch {
|
} catch {
|
||||||
// An errored or aborted stream has still ended — fall through to report
|
// An errored or aborted stream has still ended — fall through to report
|
||||||
|
|
@ -384,7 +404,12 @@ export function createFetchTracker(
|
||||||
: input instanceof URL
|
: input instanceof URL
|
||||||
? input.toString()
|
? input.toString()
|
||||||
: input.url;
|
: input.url;
|
||||||
const method = init?.method || "GET";
|
// The method decides the long-lived-stream classification below, so it
|
||||||
|
// has to be the one the request actually carries: a `Request` input
|
||||||
|
// carries its own, and HTTP methods compare case-insensitively.
|
||||||
|
const method = (
|
||||||
|
init?.method ?? (input instanceof Request ? input.method : "GET")
|
||||||
|
).toUpperCase();
|
||||||
|
|
||||||
// Extract headers, redacting sensitive values BEFORE they reach any
|
// Extract headers, redacting sensitive values BEFORE they reach any
|
||||||
// downstream sink (logger, in-memory list, persisted session storage).
|
// downstream sink (logger, in-memory list, persisted session storage).
|
||||||
|
|
|
||||||
|
|
@ -273,6 +273,7 @@ import {
|
||||||
import { SdkError, SdkErrorCode } from "@modelcontextprotocol/client";
|
import { SdkError, SdkErrorCode } from "@modelcontextprotocol/client";
|
||||||
import {
|
import {
|
||||||
annotateRequestTimeout,
|
annotateRequestTimeout,
|
||||||
|
isRequestTimeoutError,
|
||||||
type ConnectionDiagnostics,
|
type ConnectionDiagnostics,
|
||||||
type LastResponse,
|
type LastResponse,
|
||||||
type NotificationStreamState,
|
type NotificationStreamState,
|
||||||
|
|
@ -1131,11 +1132,16 @@ export class InspectorClient extends InspectorClientEventTarget {
|
||||||
};
|
};
|
||||||
this.dispatchTypedEvent("message", entry);
|
this.dispatchTypedEvent("message", entry);
|
||||||
},
|
},
|
||||||
trackSendFailure: (message: JSONRPCRequest) => {
|
trackSendFailure: (message: JSONRPCRequest, error: unknown) => {
|
||||||
// The frame never reached the wire, so it is not awaiting a response.
|
// A send that failed because the request timed out is not a send
|
||||||
// Leaving it would report a request the server never saw as
|
// that failed: the browser's remote transport awaits the response
|
||||||
// unanswered, in every timeout message and Connection Info row, until
|
// inside `send`, so its relay timeout surfaces here — the request
|
||||||
// the session resets (#2318).
|
// reached the server and went unanswered, which is the state to
|
||||||
|
// keep reporting. Every other failure means the frame never reached
|
||||||
|
// the wire, and leaving it would report a request the server never
|
||||||
|
// saw as unanswered, in every timeout message and Connection Info
|
||||||
|
// row, until the session resets (#2318).
|
||||||
|
if (isRequestTimeoutError(error)) return;
|
||||||
if (this.outstandingRequests.delete(message.id)) {
|
if (this.outstandingRequests.delete(message.id)) {
|
||||||
this.dispatchConnectionDiagnosticsChange();
|
this.dispatchConnectionDiagnosticsChange();
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -29,7 +29,11 @@ export interface MessageTrackingCallbacks {
|
||||||
* closed, the fetch failed before the frame reached the wire). Fires after
|
* closed, the fetch failed before the frame reached the wire). Fires after
|
||||||
* `trackRequest` already recorded it, so a consumer keeping "requests still
|
* `trackRequest` already recorded it, so a consumer keeping "requests still
|
||||||
* awaiting a response" can roll that entry back — a request that never went
|
* awaiting a response" can roll that entry back — a request that never went
|
||||||
* out is not unanswered (#2318).
|
* out is not unanswered (#2318). Not fired when the caller aborted the
|
||||||
|
* request through `requestSignal`: the modern transport aborts an in-flight
|
||||||
|
* request's own stream to cancel it (and the SDK does the same on a
|
||||||
|
* per-request timeout), and a request that was sent and then given up on
|
||||||
|
* is exactly the unanswered kind.
|
||||||
*/
|
*/
|
||||||
trackSendFailure?: (message: JSONRPCRequest, error: unknown) => void;
|
trackSendFailure?: (message: JSONRPCRequest, error: unknown) => void;
|
||||||
}
|
}
|
||||||
|
|
@ -106,7 +110,9 @@ export class MessageTrackingTransport implements Transport {
|
||||||
const request = message as JSONRPCRequest;
|
const request = message as JSONRPCRequest;
|
||||||
this.callbacks.trackRequest?.(request, "client");
|
this.callbacks.trackRequest?.(request, "client");
|
||||||
return this.baseTransport.send(message, options).catch((err) => {
|
return this.baseTransport.send(message, options).catch((err) => {
|
||||||
this.callbacks.trackSendFailure?.(request, err);
|
if (!options?.requestSignal?.aborted) {
|
||||||
|
this.callbacks.trackSendFailure?.(request, err);
|
||||||
|
}
|
||||||
throw err;
|
throw err;
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -55,6 +55,12 @@ export function createTransportNode(
|
||||||
interceptAuthChallenges = false,
|
interceptAuthChallenges = false,
|
||||||
onAuthChallengeObserved,
|
onAuthChallengeObserved,
|
||||||
} = options;
|
} = options;
|
||||||
|
// Any of the three tracking callbacks earns a tracker: each is optional on
|
||||||
|
// the tracker itself, and a caller that wants only stream updates (or only
|
||||||
|
// bodies) must not be silently handed the bare fetch (#2318).
|
||||||
|
const wantsFetchTracking = Boolean(
|
||||||
|
onFetchRequest || onFetchResponseBody || onFetchStreamUpdate,
|
||||||
|
);
|
||||||
|
|
||||||
// `optionsFetchFn` is the caller's whole fetch stack and already sits on top of
|
// `optionsFetchFn` is the caller's whole fetch stack and already sits on top of
|
||||||
// a proxy-aware base when one is needed — the Node clients install
|
// a proxy-aware base when one is needed — the Node clients install
|
||||||
|
|
@ -115,7 +121,7 @@ export function createTransportNode(
|
||||||
const sseFetch = configuredSseFetch
|
const sseFetch = configuredSseFetch
|
||||||
? withChallengeObserver(configuredSseFetch)
|
? withChallengeObserver(configuredSseFetch)
|
||||||
: fetchWithOptionalAuthIntercept;
|
: fetchWithOptionalAuthIntercept;
|
||||||
const trackedFetch = onFetchRequest
|
const trackedFetch = wantsFetchTracking
|
||||||
? createFetchTracker(sseFetch, {
|
? createFetchTracker(sseFetch, {
|
||||||
trackRequest: onFetchRequest,
|
trackRequest: onFetchRequest,
|
||||||
updateResponseBody: onFetchResponseBody,
|
updateResponseBody: onFetchResponseBody,
|
||||||
|
|
@ -136,7 +142,7 @@ export function createTransportNode(
|
||||||
...(headers && { headers }),
|
...(headers && { headers }),
|
||||||
};
|
};
|
||||||
|
|
||||||
const postFetch = onFetchRequest
|
const postFetch = wantsFetchTracking
|
||||||
? createFetchTracker(fetchWithOptionalAuthIntercept, {
|
? createFetchTracker(fetchWithOptionalAuthIntercept, {
|
||||||
trackRequest: onFetchRequest,
|
trackRequest: onFetchRequest,
|
||||||
updateResponseBody: onFetchResponseBody,
|
updateResponseBody: onFetchResponseBody,
|
||||||
|
|
@ -164,7 +170,7 @@ export function createTransportNode(
|
||||||
...(headers && { headers }),
|
...(headers && { headers }),
|
||||||
};
|
};
|
||||||
|
|
||||||
const transportFetch = onFetchRequest
|
const transportFetch = wantsFetchTracking
|
||||||
? createFetchTracker(fetchWithOptionalAuthIntercept, {
|
? createFetchTracker(fetchWithOptionalAuthIntercept, {
|
||||||
trackRequest: onFetchRequest,
|
trackRequest: onFetchRequest,
|
||||||
updateResponseBody: onFetchResponseBody,
|
updateResponseBody: onFetchResponseBody,
|
||||||
|
|
|
||||||
|
|
@ -11,6 +11,7 @@ import type {
|
||||||
JSONRPCMessage,
|
JSONRPCMessage,
|
||||||
MessageExtraInfo,
|
MessageExtraInfo,
|
||||||
} from "@modelcontextprotocol/client";
|
} from "@modelcontextprotocol/client";
|
||||||
|
import { SdkError, SdkErrorCode } from "@modelcontextprotocol/client";
|
||||||
import type { InspectorServerSettings, StderrLogEntry } from "../types.js";
|
import type { InspectorServerSettings, StderrLogEntry } from "../types.js";
|
||||||
import type { FetchRequestEntryBase, FetchStreamState } from "../types.js";
|
import type { FetchRequestEntryBase, FetchStreamState } from "../types.js";
|
||||||
import type {
|
import type {
|
||||||
|
|
@ -689,10 +690,17 @@ export class RemoteClientTransport implements Transport {
|
||||||
return new Promise((resolve, reject) => {
|
return new Promise((resolve, reject) => {
|
||||||
const timer = setTimeout(() => {
|
const timer = setTimeout(() => {
|
||||||
this.sseResponseWaits.delete(requestId);
|
this.sseResponseWaits.delete(requestId);
|
||||||
|
// The SDK's own per-request timeout shape, not a plain `Error`: this
|
||||||
|
// wait can expire before the SDK's timer does (its budget is the
|
||||||
|
// relay's, not the request's), and the rejection travels up through
|
||||||
|
// `Client.request`, where `InspectorClient` annotates a timeout of
|
||||||
|
// exactly this shape with the connection's state (#2318). A plain
|
||||||
|
// error would pass that decorator untouched and reach the user as a
|
||||||
|
// bare timeout on the web client alone.
|
||||||
reject(
|
reject(
|
||||||
new Error(
|
new SdkError(SdkErrorCode.RequestTimeout, "Request timed out", {
|
||||||
`Timed out waiting for MCP response on SSE (${this.sseResponseTimeoutMs}ms)`,
|
timeout: this.sseResponseTimeoutMs,
|
||||||
),
|
}),
|
||||||
);
|
);
|
||||||
}, this.sseResponseTimeoutMs);
|
}, this.sseResponseTimeoutMs);
|
||||||
this.sseResponseWaits.set(requestId, {
|
this.sseResponseWaits.set(requestId, {
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue