16415132941

Committed 21 Jul 2025 10:58AM UTC coverage: 80.343% (-1.5%) from 81.82%

Build # 16415132941

Build Type

Pull #361

github

Committed by

web-flow

Commit Message

Merge 1297ed957 into 7856bb93a

Pull Request Pull Request #361: feat: add streamable http [MCP-42]

Run Details

545 of 715 branches covered (76.22%)

Branch coverage included in aggregate %.

248 of 409 new or added lines in 10 files covered. (60.64%)

9 existing lines in 1 file now uncovered.

3109 of 3833 relevant lines covered (81.11%)

47.21 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

56.57

/src/common/sessionStore.ts

import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import logger, { LogId, McpLogger } from "./logger.js";
import { TimeoutManager } from "./timeoutManager.js";

export class SessionStore {
    private sessions: {
        [sessionId: string]: {
            mcpServer: McpServer;
            transport: StreamableHTTPServerTransport;
            abortTimeout: TimeoutManager;
            notificationTimeout: TimeoutManager;
        };
    } = {};

    constructor(
        private readonly idleTimeoutMS: number,
        private readonly notificationTimeoutMS: number
    ) {
        if (idleTimeoutMS <= 0) {
            throw new Error("idleTimeoutMS must be greater than 0");
        }
        if (notificationTimeoutMS <= 0) {
            throw new Error("notificationTimeoutMS must be greater than 0");
        }
        if (idleTimeoutMS <= notificationTimeoutMS) {
            throw new Error("idleTimeoutMS must be greater than notificationTimeoutMS");
        }
    }

    getSession(sessionId: string): StreamableHTTPServerTransport | undefined {
        this.resetTimeout(sessionId);
        return this.sessions[sessionId]?.transport;
    }

    private resetTimeout(sessionId: string): void {
        const session = this.sessions[sessionId];
        if (!session) {
            return;
        }

        session.abortTimeout.reset();

        session.notificationTimeout.reset();
    }

    private sendNotification(sessionId: string): void {
        const session = this.sessions[sessionId];
        if (!session) {
            return;
        }
        const logger = new McpLogger(session.mcpServer);
        logger.info(
            LogId.streamableHttpTransportSessionCloseNotification,
            "sessionStore",
            "Session is about to be closed due to inactivity"
        );
    }

    setSession(sessionId: string, transport: StreamableHTTPServerTransport, mcpServer: McpServer): void {
        if (this.sessions[sessionId]) {
            throw new Error(`Session ${sessionId} already exists`);
        }
        const abortTimeout = new TimeoutManager(async () => {
            const logger = new McpLogger(mcpServer);
            logger.info(
                LogId.streamableHttpTransportSessionCloseNotification,
                "sessionStore",
                "Session closed due to inactivity"
            );

            await this.closeSession(sessionId);
        }, this.idleTimeoutMS);
        const notificationTimeout = new TimeoutManager(
            () => this.sendNotification(sessionId),
            this.notificationTimeoutMS
        );
        this.sessions[sessionId] = { mcpServer, transport, abortTimeout, notificationTimeout };
    }

    async closeSession(sessionId: string, closeTransport: boolean = true): Promise<void> {
        if (!this.sessions[sessionId]) {
            throw new Error(`Session ${sessionId} not found`);
        }
        this.sessions[sessionId].abortTimeout.clear();
        this.sessions[sessionId].notificationTimeout.clear();
        if (closeTransport) {
            try {
                await this.sessions[sessionId].transport.close();
            } catch (error) {
                logger.error(
                    LogId.streamableHttpTransportSessionCloseFailure,
                    "streamableHttpTransport",
                    `Error closing transport ${sessionId}: ${error instanceof Error ? error.message : String(error)}`
                );
            }
        }
        delete this.sessions[sessionId];
    }

    async closeAllSessions(): Promise<void> {
        await Promise.all(Object.keys(this.sessions).map((sessionId) => this.closeSession(sessionId)));
    }
}

1	import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
2	import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
3	import logger, { LogId, McpLogger } from "./logger.js";	2✔
4	import { TimeoutManager } from "./timeoutManager.js";	2✔
5
6	export class SessionStore {	2✔
7	private sessions: {	2✔
8	[sessionId: string]: {
9	mcpServer: McpServer;
10	transport: StreamableHTTPServerTransport;
11	abortTimeout: TimeoutManager;
12	notificationTimeout: TimeoutManager;
13	};
14	} = {};	2✔
15
16	constructor(	2✔
17	private readonly idleTimeoutMS: number,	2✔
18	private readonly notificationTimeoutMS: number	2✔
19	) {	2✔
20	if (idleTimeoutMS <= 0) {	2!
NEW 21	throw new Error("idleTimeoutMS must be greater than 0");	×
NEW 22	}	×
23	if (notificationTimeoutMS <= 0) {	2!
NEW 24	throw new Error("notificationTimeoutMS must be greater than 0");	×
NEW 25	}	×
26	if (idleTimeoutMS <= notificationTimeoutMS) {	2!
NEW 27	throw new Error("idleTimeoutMS must be greater than notificationTimeoutMS");	×
NEW 28	}	×
29	}	2✔
30
31	getSession(sessionId: string): StreamableHTTPServerTransport \| undefined {	2✔
32	this.resetTimeout(sessionId);	6✔
33	return this.sessions[sessionId]?.transport;	6✔
34	}	6✔
35
36	private resetTimeout(sessionId: string): void {	2✔
37	const session = this.sessions[sessionId];	6✔
38	if (!session) {	6!
NEW 39	return;	×
NEW 40	}	×
41
42	session.abortTimeout.reset();	6✔
43
44	session.notificationTimeout.reset();	6✔
45	}	6✔
46
47	private sendNotification(sessionId: string): void {	2✔
NEW 48	const session = this.sessions[sessionId];	×
NEW 49	if (!session) {	×
NEW 50	return;	×
NEW 51	}	×
NEW 52	const logger = new McpLogger(session.mcpServer);	×
NEW 53	logger.info(	×
NEW 54	LogId.streamableHttpTransportSessionCloseNotification,	×
NEW 55	"sessionStore",	×
NEW 56	"Session is about to be closed due to inactivity"	×
NEW 57	);	×
NEW 58	}	×
59
60	setSession(sessionId: string, transport: StreamableHTTPServerTransport, mcpServer: McpServer): void {	2✔
61	if (this.sessions[sessionId]) {	2!
NEW 62	throw new Error(`Session ${sessionId} already exists`);	×
NEW 63	}	×
64	const abortTimeout = new TimeoutManager(async () => {	2✔
NEW 65	const logger = new McpLogger(mcpServer);	×
NEW 66	logger.info(	×
NEW 67	LogId.streamableHttpTransportSessionCloseNotification,	×
NEW 68	"sessionStore",	×
NEW 69	"Session closed due to inactivity"	×
NEW 70	);	×
71
NEW 72	await this.closeSession(sessionId);	×
73	}, this.idleTimeoutMS);	2✔
74	const notificationTimeout = new TimeoutManager(	2✔
75	() => this.sendNotification(sessionId),	2✔
76	this.notificationTimeoutMS	2✔
77	);	2✔
78	this.sessions[sessionId] = { mcpServer, transport, abortTimeout, notificationTimeout };	2✔
79	}	2✔
80
81	async closeSession(sessionId: string, closeTransport: boolean = true): Promise<void> {	2✔
82	if (!this.sessions[sessionId]) {	2!
NEW 83	throw new Error(`Session ${sessionId} not found`);	×
NEW 84	}	×
85	this.sessions[sessionId].abortTimeout.clear();	2✔
86	this.sessions[sessionId].notificationTimeout.clear();	2✔
87	if (closeTransport) {	2✔
88	try {	2✔
89	await this.sessions[sessionId].transport.close();	2✔
90	} catch (error) {	2!
NEW 91	logger.error(	×
NEW 92	LogId.streamableHttpTransportSessionCloseFailure,	×
NEW 93	"streamableHttpTransport",	×
NEW 94	`Error closing transport ${sessionId}: ${error instanceof Error ? error.message : String(error)}`	×
NEW 95	);	×
NEW 96	}	×
97	}	2✔
98	delete this.sessions[sessionId];	2✔
99	}	2✔
100
101	async closeAllSessions(): Promise<void> {	2✔
102	await Promise.all(Object.keys(this.sessions).map((sessionId) => this.closeSession(sessionId)));	2✔
103	}	2✔
104	}	2✔

mongodb-js / mongodb-mcp-server / 16415132941

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous