Allow disabling MCP client timeouts
This commit is contained in:
+3
-1
@@ -68,10 +68,12 @@ Server names must match `^[a-zA-Z0-9_.-]{1,100}$`.
|
||||
Shared fields for every transport:
|
||||
|
||||
- `enabled?: boolean` — skip this server when `false`
|
||||
- `timeout?: number` — connection timeout in milliseconds
|
||||
- `timeout?: number` — MCP request timeout in milliseconds; `0` disables client-side MCP timeouts
|
||||
- `auth?: { ... }` — auth metadata used by OMP for OAuth/API-key flows
|
||||
- `oauth?: { ... }` — explicit OAuth client settings used during auth/reauth
|
||||
|
||||
Set `OMP_MCP_TIMEOUT_MS=0` to disable the client-side timeout for every MCP server in the current process. Set it to a positive millisecond value, such as `OMP_MCP_TIMEOUT_MS=120000`, to apply one global timeout without editing each server entry.
|
||||
|
||||
### `stdio` transport
|
||||
|
||||
`stdio` is the default when `type` is omitted.
|
||||
|
||||
@@ -91,7 +91,7 @@ For each discovered server in `connectServers()`:
|
||||
- performs MCP `initialize`,
|
||||
- for HTTP/SSE, starts the optional background SSE listener before `notifications/initialized`,
|
||||
- sends `notifications/initialized`,
|
||||
- uses timeout (`config.timeout` or 30s default),
|
||||
- uses timeout (`OMP_MCP_TIMEOUT_MS`, `config.timeout`, or 30s default; `0` disables the client-side timeout),
|
||||
- closes transport on init failure.
|
||||
|
||||
### Fast startup gate + deferred fallback
|
||||
|
||||
@@ -98,8 +98,8 @@
|
||||
},
|
||||
"timeout": {
|
||||
"type": "number",
|
||||
"exclusiveMinimum": 0,
|
||||
"description": "Connection timeout in milliseconds."
|
||||
"minimum": 0,
|
||||
"description": "MCP request timeout in milliseconds. Set to 0 to disable client-side MCP timeouts."
|
||||
},
|
||||
"auth": {
|
||||
"$ref": "#/$defs/authConfig"
|
||||
|
||||
@@ -132,7 +132,7 @@ async function loadMCPServers(ctx: LoadContext): Promise<LoadResult<MCPServer>>
|
||||
if (serverConfig.timeout === undefined || serverConfig.timeout === null) {
|
||||
timeout = undefined;
|
||||
} else if (typeof serverConfig.timeout === "number") {
|
||||
if (Number.isFinite(serverConfig.timeout) && serverConfig.timeout > 0) {
|
||||
if (Number.isFinite(serverConfig.timeout) && serverConfig.timeout >= 0) {
|
||||
timeout = serverConfig.timeout;
|
||||
} else {
|
||||
logger.warn(`MCP server "${serverName}": invalid timeout ${serverConfig.timeout}, ignoring`);
|
||||
@@ -140,7 +140,7 @@ async function loadMCPServers(ctx: LoadContext): Promise<LoadResult<MCPServer>>
|
||||
}
|
||||
} else if (typeof serverConfig.timeout === "string") {
|
||||
const parsed = Number(serverConfig.timeout);
|
||||
if (Number.isFinite(parsed) && parsed > 0) {
|
||||
if (Number.isFinite(parsed) && parsed >= 0) {
|
||||
timeout = parsed;
|
||||
} else {
|
||||
logger.warn(`MCP server "${serverName}": invalid timeout "${serverConfig.timeout}", ignoring`);
|
||||
|
||||
@@ -74,7 +74,7 @@ function transformMCPConfig(config: MCPConfigFile, source: SourceMeta): MCPServe
|
||||
if (
|
||||
typeof serverConfig.timeout === "number" &&
|
||||
Number.isFinite(serverConfig.timeout) &&
|
||||
serverConfig.timeout > 0
|
||||
serverConfig.timeout >= 0
|
||||
) {
|
||||
timeout = serverConfig.timeout;
|
||||
} else {
|
||||
|
||||
@@ -6,6 +6,7 @@
|
||||
import * as path from "node:path";
|
||||
import * as url from "node:url";
|
||||
import { getProjectDir, logger, withTimeout } from "@oh-my-pi/pi-utils";
|
||||
import { describeMCPTimeout, isMCPTimeoutEnabled, resolveMCPTimeoutMs } from "./timeout";
|
||||
import { createHttpTransport } from "./transports/http";
|
||||
import { createStdioTransport } from "./transports/stdio";
|
||||
import type {
|
||||
@@ -39,9 +40,6 @@ import type {
|
||||
/** MCP protocol version we support */
|
||||
const PROTOCOL_VERSION = "2025-03-26";
|
||||
|
||||
/** Default connection timeout in ms */
|
||||
const CONNECTION_TIMEOUT_MS = 30_000;
|
||||
|
||||
/** Client info sent during initialization */
|
||||
const CLIENT_INFO = {
|
||||
name: "omp-coding-agent",
|
||||
@@ -128,7 +126,8 @@ async function initializeConnection(
|
||||
|
||||
/**
|
||||
* Connect to an MCP server.
|
||||
* Has a 30 second timeout to prevent blocking startup.
|
||||
* Has a 30 second timeout by default to prevent blocking startup.
|
||||
* Set OMP_MCP_TIMEOUT_MS=0 to disable MCP client-side timeouts.
|
||||
*/
|
||||
export async function connectToServer(
|
||||
name: string,
|
||||
@@ -139,7 +138,7 @@ export async function connectToServer(
|
||||
onRequest?: (method: string, params: unknown) => Promise<unknown>;
|
||||
},
|
||||
): Promise<MCPServerConnection> {
|
||||
const timeoutMs = config.timeout ?? CONNECTION_TIMEOUT_MS;
|
||||
const timeoutMs = resolveMCPTimeoutMs(config.timeout);
|
||||
let transport: MCPTransport | undefined;
|
||||
|
||||
const connect = async (): Promise<MCPServerConnection> => {
|
||||
@@ -180,10 +179,13 @@ export async function connectToServer(
|
||||
};
|
||||
|
||||
try {
|
||||
if (!isMCPTimeoutEnabled(timeoutMs)) {
|
||||
return await connect();
|
||||
}
|
||||
return await withTimeout(
|
||||
connect(),
|
||||
timeoutMs,
|
||||
`Connection to MCP server "${name}" timed out after ${timeoutMs}ms`,
|
||||
`Connection to MCP server "${name}" timed out after ${describeMCPTimeout(timeoutMs)}`,
|
||||
options?.signal,
|
||||
);
|
||||
} catch (error) {
|
||||
|
||||
@@ -0,0 +1,54 @@
|
||||
const DEFAULT_MCP_TIMEOUT_MS = 30_000;
|
||||
const MCP_TIMEOUT_ENV = "OMP_MCP_TIMEOUT_MS";
|
||||
|
||||
let neverAbortController: AbortController | undefined;
|
||||
|
||||
export function resolveMCPTimeoutMs(configTimeout?: number): number {
|
||||
const raw = Bun.env[MCP_TIMEOUT_ENV]?.trim();
|
||||
if (raw) {
|
||||
const value = Number(raw);
|
||||
if (Number.isFinite(value)) return value;
|
||||
}
|
||||
return configTimeout ?? DEFAULT_MCP_TIMEOUT_MS;
|
||||
}
|
||||
|
||||
export function isMCPTimeoutEnabled(timeoutMs: number): boolean {
|
||||
return timeoutMs > 0;
|
||||
}
|
||||
|
||||
export function describeMCPTimeout(timeoutMs: number): string {
|
||||
return isMCPTimeoutEnabled(timeoutMs) ? `${timeoutMs}ms` : "disabled";
|
||||
}
|
||||
|
||||
export function getNeverAbortSignal(): AbortSignal {
|
||||
neverAbortController ??= new AbortController();
|
||||
return neverAbortController.signal;
|
||||
}
|
||||
|
||||
export function createMCPTimeout(
|
||||
timeoutMs: number,
|
||||
signal?: AbortSignal,
|
||||
): {
|
||||
signal?: AbortSignal;
|
||||
clear: () => void;
|
||||
isTimeoutAbort: (error: unknown) => boolean;
|
||||
} {
|
||||
if (!isMCPTimeoutEnabled(timeoutMs)) {
|
||||
return {
|
||||
signal,
|
||||
clear: () => {},
|
||||
isTimeoutAbort: () => false,
|
||||
};
|
||||
}
|
||||
|
||||
const abortController = new AbortController();
|
||||
const timeoutId = setTimeout(() => abortController.abort(), timeoutMs);
|
||||
const operationSignal = signal ? AbortSignal.any([signal, abortController.signal]) : abortController.signal;
|
||||
|
||||
return {
|
||||
signal: operationSignal,
|
||||
clear: () => clearTimeout(timeoutId),
|
||||
isTimeoutAbort: error =>
|
||||
error instanceof Error && error.name === "AbortError" && abortController.signal.aborted && !signal?.aborted,
|
||||
};
|
||||
}
|
||||
@@ -16,6 +16,7 @@ import type {
|
||||
MCPTransport,
|
||||
} from "../../mcp/types";
|
||||
import { toJsonRpcError } from "../../mcp/types";
|
||||
import { createMCPTimeout, getNeverAbortSignal, resolveMCPTimeoutMs } from "../timeout";
|
||||
|
||||
/**
|
||||
* HTTP transport for MCP servers.
|
||||
@@ -180,23 +181,18 @@ export class HttpTransport implements MCPTransport {
|
||||
headers["Mcp-Session-Id"] = this.#sessionId;
|
||||
}
|
||||
|
||||
// Create AbortController for timeout
|
||||
const timeout = this.config.timeout ?? 30000;
|
||||
const abortController = new AbortController();
|
||||
const timeoutId = setTimeout(() => abortController.abort(), timeout);
|
||||
const operationSignal = options?.signal
|
||||
? AbortSignal.any([options.signal, abortController.signal])
|
||||
: abortController.signal;
|
||||
const timeout = resolveMCPTimeoutMs(this.config.timeout);
|
||||
const operation = createMCPTimeout(timeout, options?.signal);
|
||||
|
||||
try {
|
||||
const response = await fetch(this.config.url, {
|
||||
method: "POST",
|
||||
headers,
|
||||
body: JSON.stringify(body),
|
||||
signal: operationSignal,
|
||||
signal: operation.signal,
|
||||
});
|
||||
|
||||
clearTimeout(timeoutId);
|
||||
operation.clear();
|
||||
|
||||
// Check for session ID in response
|
||||
const newSessionId = response.headers.get("Mcp-Session-Id");
|
||||
@@ -234,11 +230,8 @@ export class HttpTransport implements MCPTransport {
|
||||
|
||||
return result.result as T;
|
||||
} catch (error) {
|
||||
clearTimeout(timeoutId);
|
||||
if (error instanceof Error && error.name === "AbortError") {
|
||||
if (options?.signal?.aborted) {
|
||||
throw error;
|
||||
}
|
||||
operation.clear();
|
||||
if (operation.isTimeoutAbort(error)) {
|
||||
throw new Error(`Request timeout after ${timeout}ms`);
|
||||
}
|
||||
throw error;
|
||||
@@ -250,12 +243,9 @@ export class HttpTransport implements MCPTransport {
|
||||
throw new Error("No response body");
|
||||
}
|
||||
|
||||
const timeout = this.config.timeout ?? 30000;
|
||||
const abortController = new AbortController();
|
||||
const timeoutId = setTimeout(() => abortController.abort(), timeout);
|
||||
const operationSignal = options?.signal
|
||||
? AbortSignal.any([options.signal, abortController.signal])
|
||||
: abortController.signal;
|
||||
const timeout = resolveMCPTimeoutMs(this.config.timeout);
|
||||
const operation = createMCPTimeout(timeout, options?.signal);
|
||||
const signal = operation.signal ?? getNeverAbortSignal();
|
||||
|
||||
const { promise, resolve, reject } = Promise.withResolvers<T>();
|
||||
let captured = false;
|
||||
@@ -268,7 +258,7 @@ export class HttpTransport implements MCPTransport {
|
||||
// controller", so we must not exit the loop early.
|
||||
const drain = async (): Promise<void> => {
|
||||
try {
|
||||
for await (const raw of readSseJson<JsonRpcMessage | JsonRpcMessage[]>(response.body!, operationSignal)) {
|
||||
for await (const raw of readSseJson<JsonRpcMessage | JsonRpcMessage[]>(response.body!, signal)) {
|
||||
const messages = Array.isArray(raw) ? raw : [raw];
|
||||
for (const message of messages) {
|
||||
if (
|
||||
@@ -278,7 +268,7 @@ export class HttpTransport implements MCPTransport {
|
||||
("result" in message || "error" in message)
|
||||
) {
|
||||
captured = true;
|
||||
clearTimeout(timeoutId);
|
||||
operation.clear();
|
||||
if (message.error) {
|
||||
reject(new Error(`MCP error ${message.error.code}: ${message.error.message}`));
|
||||
} else {
|
||||
@@ -295,17 +285,13 @@ export class HttpTransport implements MCPTransport {
|
||||
}
|
||||
} catch (error) {
|
||||
if (captured) return;
|
||||
if (error instanceof Error && error.name === "AbortError") {
|
||||
if (options?.signal?.aborted) {
|
||||
reject(error);
|
||||
} else {
|
||||
reject(new Error(`SSE response timeout after ${timeout}ms`));
|
||||
}
|
||||
if (operation.isTimeoutAbort(error)) {
|
||||
reject(new Error(`SSE response timeout after ${timeout}ms`));
|
||||
} else {
|
||||
reject(error as Error);
|
||||
}
|
||||
} finally {
|
||||
clearTimeout(timeoutId);
|
||||
operation.clear();
|
||||
}
|
||||
};
|
||||
|
||||
@@ -340,13 +326,16 @@ export class HttpTransport implements MCPTransport {
|
||||
if (this.#sessionId) {
|
||||
headers["Mcp-Session-Id"] = this.#sessionId;
|
||||
}
|
||||
const timeout = resolveMCPTimeoutMs(this.config.timeout);
|
||||
let operation = createMCPTimeout(timeout);
|
||||
try {
|
||||
const resp = await fetch(this.config.url, {
|
||||
method: "POST",
|
||||
headers,
|
||||
body: JSON.stringify(body),
|
||||
signal: AbortSignal.timeout(this.config.timeout ?? 30000),
|
||||
signal: operation.signal,
|
||||
});
|
||||
operation.clear();
|
||||
// Retry once on auth failure if onAuthError is wired
|
||||
if (this.onAuthError && (resp.status === 401 || resp.status === 403)) {
|
||||
await resp.body?.cancel();
|
||||
@@ -355,18 +344,21 @@ export class HttpTransport implements MCPTransport {
|
||||
this.config.headers ??= {};
|
||||
Object.assign(this.config.headers, newHeaders);
|
||||
Object.assign(headers, newHeaders);
|
||||
operation = createMCPTimeout(timeout);
|
||||
const retry = await fetch(this.config.url, {
|
||||
method: "POST",
|
||||
headers,
|
||||
body: JSON.stringify(body),
|
||||
signal: AbortSignal.timeout(this.config.timeout ?? 30000),
|
||||
signal: operation.signal,
|
||||
});
|
||||
operation.clear();
|
||||
await retry.body?.cancel();
|
||||
return;
|
||||
}
|
||||
}
|
||||
await resp.body?.cancel();
|
||||
} catch {
|
||||
operation.clear();
|
||||
// Best-effort response delivery — server may have disconnected
|
||||
}
|
||||
}
|
||||
@@ -392,20 +384,18 @@ export class HttpTransport implements MCPTransport {
|
||||
headers["Mcp-Session-Id"] = this.#sessionId;
|
||||
}
|
||||
|
||||
// Create AbortController for timeout
|
||||
const timeout = this.config.timeout ?? 30000;
|
||||
const abortController = new AbortController();
|
||||
const timeoutId = setTimeout(() => abortController.abort(), timeout);
|
||||
const timeout = resolveMCPTimeoutMs(this.config.timeout);
|
||||
const operation = createMCPTimeout(timeout);
|
||||
|
||||
try {
|
||||
const response = await fetch(this.config.url, {
|
||||
method: "POST",
|
||||
headers,
|
||||
body: JSON.stringify(body),
|
||||
signal: abortController.signal,
|
||||
signal: operation.signal,
|
||||
});
|
||||
|
||||
clearTimeout(timeoutId);
|
||||
operation.clear();
|
||||
|
||||
// 202 Accepted is success for notifications
|
||||
if (!response.ok && response.status !== 202) {
|
||||
@@ -417,15 +407,20 @@ export class HttpTransport implements MCPTransport {
|
||||
// on the notification response (MCP Streamable HTTP spec). Read them.
|
||||
const contentType = response.headers.get("Content-Type") ?? "";
|
||||
if (contentType.includes("text/event-stream") && response.body) {
|
||||
// Use the SSE connection's signal if available, otherwise read until stream ends
|
||||
const signal = this.#sseConnection?.signal ?? AbortSignal.timeout(this.config.timeout ?? 30000);
|
||||
void this.#readSSEStream(response.body, signal);
|
||||
// Use the SSE connection's signal if available; otherwise keep the existing finite read timeout.
|
||||
if (this.#sseConnection) {
|
||||
void this.#readSSEStream(response.body, this.#sseConnection.signal);
|
||||
} else {
|
||||
const readOperation = createMCPTimeout(timeout);
|
||||
const signal = readOperation.signal ?? getNeverAbortSignal();
|
||||
void this.#readSSEStream(response.body, signal).finally(() => readOperation.clear());
|
||||
}
|
||||
} else {
|
||||
await response.body?.cancel();
|
||||
}
|
||||
} catch (error) {
|
||||
clearTimeout(timeoutId);
|
||||
if (error instanceof Error && error.name === "AbortError") {
|
||||
operation.clear();
|
||||
if (operation.isTimeoutAbort(error)) {
|
||||
throw new Error(`Notify timeout after ${timeout}ms`);
|
||||
}
|
||||
throw error;
|
||||
@@ -444,8 +439,9 @@ export class HttpTransport implements MCPTransport {
|
||||
|
||||
// Send session termination if we have a session
|
||||
if (this.#sessionId) {
|
||||
const timeout = resolveMCPTimeoutMs(this.config.timeout);
|
||||
const operation = createMCPTimeout(timeout);
|
||||
try {
|
||||
const timeout = this.config.timeout ?? 30000;
|
||||
const headers: Record<string, string> = {
|
||||
...this.config.headers,
|
||||
"Mcp-Session-Id": this.#sessionId,
|
||||
@@ -454,9 +450,11 @@ export class HttpTransport implements MCPTransport {
|
||||
await fetch(this.config.url, {
|
||||
method: "DELETE",
|
||||
headers,
|
||||
signal: AbortSignal.timeout(timeout),
|
||||
signal: operation.signal,
|
||||
});
|
||||
operation.clear();
|
||||
} catch {
|
||||
operation.clear();
|
||||
// Ignore termination errors
|
||||
}
|
||||
this.#sessionId = null;
|
||||
|
||||
@@ -17,6 +17,7 @@ import type {
|
||||
MCPTransport,
|
||||
} from "../../mcp/types";
|
||||
import { toJsonRpcError } from "../../mcp/types";
|
||||
import { isMCPTimeoutEnabled, resolveMCPTimeoutMs } from "../timeout";
|
||||
|
||||
/**
|
||||
* Stdio transport for MCP servers.
|
||||
@@ -208,7 +209,7 @@ export class StdioTransport implements MCPTransport {
|
||||
params: params ?? {},
|
||||
};
|
||||
|
||||
const timeout = this.config.timeout ?? 30000;
|
||||
const timeout = resolveMCPTimeoutMs(this.config.timeout);
|
||||
const signal = options?.signal;
|
||||
|
||||
if (signal?.aborted) {
|
||||
@@ -254,10 +255,12 @@ export class StdioTransport implements MCPTransport {
|
||||
},
|
||||
});
|
||||
|
||||
timer = setTimeout(() => {
|
||||
cleanup();
|
||||
reject(new Error(`Request timeout after ${timeout}ms`));
|
||||
}, timeout);
|
||||
if (isMCPTimeoutEnabled(timeout)) {
|
||||
timer = setTimeout(() => {
|
||||
cleanup();
|
||||
reject(new Error(`Request timeout after ${timeout}ms`));
|
||||
}, timeout);
|
||||
}
|
||||
|
||||
const message = `${JSON.stringify(request)}\n`;
|
||||
try {
|
||||
|
||||
@@ -61,7 +61,7 @@ export interface MCPAuthConfig {
|
||||
interface MCPServerConfigBase {
|
||||
/** Whether this server is enabled (default: true) */
|
||||
enabled?: boolean;
|
||||
/** Connection timeout in milliseconds (default: 30000) */
|
||||
/** MCP request timeout in milliseconds (default: 30000, 0 to disable) */
|
||||
timeout?: number;
|
||||
/** Authentication configuration (optional) */
|
||||
auth?: MCPAuthConfig;
|
||||
|
||||
@@ -0,0 +1,40 @@
|
||||
import { afterEach, describe, expect, test } from "bun:test";
|
||||
import { isMCPTimeoutEnabled, resolveMCPTimeoutMs } from "../src/mcp/timeout";
|
||||
|
||||
const ORIGINAL_TIMEOUT = process.env.OMP_MCP_TIMEOUT_MS;
|
||||
|
||||
afterEach(() => {
|
||||
if (ORIGINAL_TIMEOUT === undefined) {
|
||||
delete process.env.OMP_MCP_TIMEOUT_MS;
|
||||
} else {
|
||||
process.env.OMP_MCP_TIMEOUT_MS = ORIGINAL_TIMEOUT;
|
||||
}
|
||||
});
|
||||
|
||||
describe("MCP timeout configuration", () => {
|
||||
test("uses the default timeout when no config or env override is set", () => {
|
||||
delete process.env.OMP_MCP_TIMEOUT_MS;
|
||||
|
||||
expect(resolveMCPTimeoutMs()).toBe(30_000);
|
||||
});
|
||||
|
||||
test("uses per-server timeout when env override is unset", () => {
|
||||
delete process.env.OMP_MCP_TIMEOUT_MS;
|
||||
|
||||
expect(resolveMCPTimeoutMs(120_000)).toBe(120_000);
|
||||
});
|
||||
|
||||
test("allows the env override to disable MCP client-side timeouts", () => {
|
||||
process.env.OMP_MCP_TIMEOUT_MS = "0";
|
||||
|
||||
const timeout = resolveMCPTimeoutMs(30_000);
|
||||
expect(timeout).toBe(0);
|
||||
expect(isMCPTimeoutEnabled(timeout)).toBe(false);
|
||||
});
|
||||
|
||||
test("allows the env override to set one timeout for every server", () => {
|
||||
process.env.OMP_MCP_TIMEOUT_MS = "180000";
|
||||
|
||||
expect(resolveMCPTimeoutMs(30_000)).toBe(180_000);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user