Make the agent MCP server stateless to fix session leak
The agent-facing MCP server (/mcp/agents) ran in stateful session mode, keeping a StreamableHTTPServerTransport plus its full Server and Zod tool schemas alive per agent connection, keyed by session id. The map was only pruned on the SDK's onclose/onsessionclosed callbacks, which fire only on a clean DELETE. Agents that are killed or exit without one never get reaped, so dead sessions accumulate for the life of the daemon (observed ~77MB across 132 stale transports in a heap snapshot). Listing and calling tools needs no cross-request state, so each request now builds a fresh server + transport and tears both down when the response closes. Per-request cost is ~0.6-0.9ms and ~760KB of short-lived allocation; nothing is retained between requests. GET and DELETE return 405, which the MCP client tolerates (it never receives a session id).
This commit is contained in:
@@ -6,7 +6,6 @@ import { randomUUID } from "node:crypto";
|
||||
import { hostname as getHostname } from "node:os";
|
||||
import path from "node:path";
|
||||
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
|
||||
import { isInitializeRequest } from "@modelcontextprotocol/sdk/types.js";
|
||||
import type { Logger } from "pino";
|
||||
import { z } from "zod";
|
||||
import { createBranchChangeRouteHandler } from "./script-route-branch-handler.js";
|
||||
@@ -147,8 +146,6 @@ import {
|
||||
type DaemonAuthConfig,
|
||||
} from "./auth.js";
|
||||
|
||||
type AgentMcpTransportMap = Map<string, StreamableHTTPServerTransport>;
|
||||
|
||||
const MAX_MCP_DEBUG_BATCH_ITEMS = 10;
|
||||
const REDACTED_LOG_VALUE = "[redacted]";
|
||||
const DOWNLOAD_OPEN_FLAGS =
|
||||
@@ -821,9 +818,8 @@ export async function createPaseoDaemon(
|
||||
let agentMcpBaseUrl: string | null = null;
|
||||
if (mcpEnabled) {
|
||||
const agentMcpRoute = "/mcp/agents";
|
||||
const agentMcpTransports: AgentMcpTransportMap = new Map();
|
||||
|
||||
const createAgentMcpTransport = async (callerAgentId?: string) => {
|
||||
const createAgentMcpSession = async (callerAgentId?: string) => {
|
||||
const agentMcpServer = await createAgentMcpServer({
|
||||
agentManager,
|
||||
agentStorage,
|
||||
@@ -895,34 +891,25 @@ export async function createPaseoDaemon(
|
||||
logger,
|
||||
});
|
||||
|
||||
// Stateless mode: each HTTP request builds a fresh server + transport that is
|
||||
// torn down when the response closes, so no per-session state is retained between
|
||||
// requests. The agent control plane only lists and calls tools, neither of which
|
||||
// needs cross-request state, so sessions would only pin memory for the life of the
|
||||
// daemon (agents that exit without a clean DELETE never get reaped).
|
||||
const transport = new StreamableHTTPServerTransport({
|
||||
sessionIdGenerator: () => randomUUID(),
|
||||
onsessioninitialized: (sessionId) => {
|
||||
agentMcpTransports.set(sessionId, transport);
|
||||
logger.debug({ sessionId }, "Agent MCP session initialized");
|
||||
},
|
||||
onsessionclosed: (sessionId) => {
|
||||
agentMcpTransports.delete(sessionId);
|
||||
logger.debug({ sessionId }, "Agent MCP session closed");
|
||||
},
|
||||
sessionIdGenerator: undefined,
|
||||
// NOTE: We enforce a Vite-like host allowlist at the app/websocket layer.
|
||||
// StreamableHTTPServerTransport's built-in check requires exact Host header matches.
|
||||
enableDnsRebindingProtection: false,
|
||||
});
|
||||
|
||||
Object.assign(transport, {
|
||||
onclose: () => {
|
||||
if (transport.sessionId) {
|
||||
agentMcpTransports.delete(transport.sessionId);
|
||||
}
|
||||
},
|
||||
onerror: (err: Error) => {
|
||||
logger.error({ err }, "Agent MCP transport error");
|
||||
},
|
||||
});
|
||||
|
||||
await agentMcpServer.connect(transport);
|
||||
return transport;
|
||||
return { server: agentMcpServer, transport };
|
||||
};
|
||||
|
||||
const runAgentMcpRequest = async (
|
||||
@@ -956,41 +943,32 @@ export async function createPaseoDaemon(
|
||||
);
|
||||
}
|
||||
try {
|
||||
const sessionId = req.header("mcp-session-id");
|
||||
let transport = sessionId ? agentMcpTransports.get(sessionId) : undefined;
|
||||
|
||||
if (!transport) {
|
||||
if (req.method !== "POST") {
|
||||
res.status(400).json({
|
||||
jsonrpc: "2.0",
|
||||
error: {
|
||||
code: -32000,
|
||||
message: "Missing or invalid MCP session",
|
||||
},
|
||||
id: null,
|
||||
});
|
||||
return;
|
||||
}
|
||||
if (!isInitializeRequest(req.body)) {
|
||||
res.status(400).json({
|
||||
jsonrpc: "2.0",
|
||||
error: {
|
||||
code: -32000,
|
||||
message: "Initialization request expected",
|
||||
},
|
||||
id: null,
|
||||
});
|
||||
return;
|
||||
}
|
||||
const callerAgentIdRaw = req.query.callerAgentId;
|
||||
let callerAgentId: string | undefined;
|
||||
if (typeof callerAgentIdRaw === "string") {
|
||||
callerAgentId = callerAgentIdRaw;
|
||||
} else if (Array.isArray(callerAgentIdRaw) && typeof callerAgentIdRaw[0] === "string") {
|
||||
callerAgentId = callerAgentIdRaw[0];
|
||||
}
|
||||
transport = await createAgentMcpTransport(callerAgentId);
|
||||
// Stateless: GET (standalone SSE) and DELETE (session termination) have no
|
||||
// meaning without sessions. The MCP client tolerates 405 on the GET stream
|
||||
// and never issues a DELETE because it is never handed a session id.
|
||||
if (req.method !== "POST") {
|
||||
res.status(405).json({
|
||||
jsonrpc: "2.0",
|
||||
error: {
|
||||
code: -32000,
|
||||
message: "Method not allowed",
|
||||
},
|
||||
id: null,
|
||||
});
|
||||
return;
|
||||
}
|
||||
const callerAgentIdRaw = req.query.callerAgentId;
|
||||
let callerAgentId: string | undefined;
|
||||
if (typeof callerAgentIdRaw === "string") {
|
||||
callerAgentId = callerAgentIdRaw;
|
||||
} else if (Array.isArray(callerAgentIdRaw) && typeof callerAgentIdRaw[0] === "string") {
|
||||
callerAgentId = callerAgentIdRaw[0];
|
||||
}
|
||||
const { server, transport } = await createAgentMcpSession(callerAgentId);
|
||||
res.on("close", () => {
|
||||
void transport.close();
|
||||
void server.close();
|
||||
});
|
||||
|
||||
await transport.handleRequest(
|
||||
req as unknown as IncomingMessage,
|
||||
|
||||
Reference in New Issue
Block a user