Skip to content

fix(mcp): wait for client event stream before server->client requests #41559

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pavelfeldman merged 1 commit into from
Jul 1, 2026
+26 −14
Merged

fix(mcp): wait for client event stream before server->client requests #41559

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion packages/playwright-core/src/tools/mcp/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ export async function createConnection(userConfig: Config = {}, contextGetter?:
},
disposed: async () => { }
};
return createServer('api', packageJSON.version, backendFactory, false);
return createServer('api', packageJSON.version, backendFactory, Promise.resolve(), false);
}

class SimpleBrowser {
Expand Down
1 change: 1 addition & 0 deletions packages/playwright-core/src/tools/utils/mcp/DEPS.list
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
[*]
@utils/**
@isomorphic/**
node_modules/@modelcontextprotocol/sdk/server/index.js
node_modules/@modelcontextprotocol/sdk/server/sse.js
node_modules/@modelcontextprotocol/sdk/server/stdio.js
Expand Down
24 changes: 17 additions & 7 deletions packages/playwright-core/src/tools/utils/mcp/http.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ import { SSEServerTransport } from '@modelcontextprotocol/sdk/server/sse.js';
import { StreamableHTTPServerTransport } from '@modelcontextprotocol/sdk/server/streamableHttp.js';
import { urlHostFromAddress } from '@utils/httpServer';
import { createHttpServer, startHttpServer } from '@utils/network';
import { ManualPromise } from '@isomorphic/manualPromise';

import * as mcpServer from './server';

Expand Down Expand Up @@ -123,7 +124,7 @@ async function handleSSE(serverBackendFactory: ServerBackendFactory, req: http.I
const transport = new SSEServerTransport('/sse', res);
sessions.set(transport.sessionId, transport);
testDebug(`create SSE session`);
await mcpServer.connect(serverBackendFactory, transport, false);
await mcpServer.connect(serverBackendFactory, transport, Promise.resolve(), false);
res.on('close', () => {
testDebug(`delete SSE session`);
sessions.delete(transport.sessionId);
Expand All @@ -135,25 +136,34 @@ async function handleSSE(serverBackendFactory: ServerBackendFactory, req: http.I
res.end('Method not allowed');
}

async function handleStreamable(serverBackendFactory: ServerBackendFactory, req: http.IncomingMessage, res: http.ServerResponse, sessions: Map<string, StreamableHTTPServerTransportType>) {
async function handleStreamable(serverBackendFactory: ServerBackendFactory, req: http.IncomingMessage, res: http.ServerResponse, sessions: Map<string, { transport: StreamableHTTPServerTransportType, transportInitialized: ManualPromise<void> }>) {
const sessionId = req.headers['mcp-session-id'] as string | undefined;
if (sessionId) {
const transport = sessions.get(sessionId);
if (!transport) {
const sessionInfo = sessions.get(sessionId);
if (!sessionInfo) {
res.statusCode = 404;
res.end('Session not found');
return;
}
return await transport.handleRequest(req, res);
if (req.method === 'GET') {
// As per spec, GET is for the event stream only, when we see it consider transport bidirectionally ready.
const streamResponse = sessionInfo.transport.handleRequest(req, res);
sessionInfo.transportInitialized.resolve();
return streamResponse;
}
return sessionInfo.transport.handleRequest(req, res);
}

if (req.method === 'POST') {
const transport = new StreamableHTTPServerTransport({
sessionIdGenerator: () => crypto.randomUUID(),
onsessioninitialized: async sessionId => {
testDebug(`create http session`);
await mcpServer.connect(serverBackendFactory, transport, true);
sessions.set(sessionId, transport);
const sessionInfo = { transport, transportInitialized: new ManualPromise() };
// Only give the client 5 seconds to reach for the event stream.
setTimeout(() => sessionInfo.transportInitialized.resolve(), 5000);
sessions.set(sessionId, sessionInfo);
await mcpServer.connect(serverBackendFactory, sessionInfo.transport, sessionInfo.transportInitialized, true);
}
});

Expand Down
13 changes: 7 additions & 6 deletions packages/playwright-core/src/tools/utils/mcp/server.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,12 +75,12 @@ export type ServerBackendFactory = {
disposed: (backend: ServerBackend) => Promise<void>;
};

export async function connect(factory: ServerBackendFactory, transport: Transport, runHeartbeat: boolean) {
const server = createServer(factory.name, factory.version, factory, runHeartbeat);
export async function connect(factory: ServerBackendFactory, transport: Transport, transportInitialized: Promise<void>, runHeartbeat: boolean) {
const server = createServer(factory.name, factory.version, factory, transportInitialized, runHeartbeat);
await server.connect(transport);
}

export function createServer(name: string, version: string, factory: ServerBackendFactory, runHeartbeat: boolean): ServerType {
export function createServer(name: string, version: string, factory: ServerBackendFactory, transportInitialized: Promise<void>, runHeartbeat: boolean): ServerType {
const server = new Server({ name, version }, {
capabilities: {
tools: {},
Expand All @@ -102,7 +102,7 @@ export function createServer(name: string, version: string, factory: ServerBacke

try {
if (!backendPromise) {
backendPromise = initializeServer(server, factory, runHeartbeat).catch(e => {
backendPromise = initializeServer(server, factory, transportInitialized, runHeartbeat).catch(e => {
backendPromise = undefined;
throw e;
});
Expand All @@ -129,10 +129,11 @@ export function createServer(name: string, version: string, factory: ServerBacke
return server;
}

const initializeServer = async (server: ServerType, factory: ServerBackendFactory, runHeartbeat: boolean): Promise<ServerBackend> => {
const initializeServer = async (server: ServerType, factory: ServerBackendFactory, transportInitialized: Promise<void>, runHeartbeat: boolean): Promise<ServerBackend> => {
const capabilities = server.getClientCapabilities();
let clientRoots: Root[] = [];
if (capabilities?.roots) {
await transportInitialized;

@yury-s yury-s Jun 30, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should it just resolve to false in this case, so that we don't need to send listRoots at all?

@pavelfeldman pavelfeldman Jul 1, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm trying to be extra conservative here: wait for up to 5 seconds, relative to what was happening before the change, to not affect customers.

const { roots } = await server.listRoots().catch(e => {
serverDebug(e);
return { roots: [] };
Expand Down Expand Up @@ -196,7 +197,7 @@ export async function start(serverBackendFactory: ServerBackendFactory, options:
// The SDK's StdioServerTransport doesn't detect peer disconnect — it never listens for stdin
// end-of-stream. Wire it up so callTool requests can be cancelled when the client goes away.
process.stdin.on('end', () => void transport.close());
await connect(serverBackendFactory, transport, false);
await connect(serverBackendFactory, transport, Promise.resolve(), false);
return;
}

Expand Down
Loading