diff --git a/integrations/mcp-server/README.md b/integrations/mcp-server/README.md index 422efae..c261125 100644 --- a/integrations/mcp-server/README.md +++ b/integrations/mcp-server/README.md @@ -16,6 +16,10 @@ node integrations/mcp-server/src/index.mjs Set `ANDROID_SERIAL` to target a specific ADB device or emulator. Set `OPENPHONE_DRY_RUN=1` for parser/protocol tests without ADB. +The server speaks the MCP stdio transport: newline-delimited JSON-RPC (one +message per line, no `Content-Length` headers) and negotiates +`protocolVersion` during `initialize` (latest supported: `2025-06-18`). + The same ADB transport works with the OpenPhone SDK phone emulator: ```sh diff --git a/integrations/mcp-server/src/index.mjs b/integrations/mcp-server/src/index.mjs index 37306e6..cc6a94f 100644 --- a/integrations/mcp-server/src/index.mjs +++ b/integrations/mcp-server/src/index.mjs @@ -8,11 +8,19 @@ import { textContent, } from "../runtime/protocol/openphone-runtime-tools.mjs"; -const PROTOCOL_VERSION = "2025-11-25"; +export const SUPPORTED_PROTOCOL_VERSIONS = ["2025-06-18"]; +const LATEST_PROTOCOL_VERSION = SUPPORTED_PROTOCOL_VERSIONS[0]; const SERVER_INFO = { name: "openphone-runtime", version: "0.1.0" }; const commands = loadCommands(); +export function negotiateProtocolVersion(requested) { + if (SUPPORTED_PROTOCOL_VERSIONS.includes(requested)) { + return requested; + } + return LATEST_PROTOCOL_VERSION; +} + export async function handleRequest(message, context = {}) { if (!message || typeof message !== "object") { return errorResponse(null, -32600, "Invalid Request"); @@ -26,7 +34,7 @@ export async function handleRequest(message, context = {}) { switch (message.method) { case "initialize": return resultResponse(id, { - protocolVersion: PROTOCOL_VERSION, + protocolVersion: negotiateProtocolVersion(message.params?.protocolVersion), capabilities: { tools: { listChanged: false }, }, @@ -48,16 +56,39 @@ export async function handleRequest(message, context = {}) { export async function serve(options = {}) { const transport = options.transport ?? new OpenPhoneAdbTransport(); - let buffer = Buffer.alloc(0); - process.stdin.on("data", async (chunk) => { - buffer = Buffer.concat([buffer, chunk]); - const parsed = parseMessages(buffer); - buffer = parsed.remaining; - for (const message of parsed.messages) { - const response = await handleRequest(message, { transport }); - if (response) { - writeMessage(response); + let buffer = ""; + let queue = Promise.resolve(); + process.stdin.setEncoding("utf8"); + process.stdin.on("data", (chunk) => { + // Buffer mutation must stay synchronous: extract every complete line + // before awaiting anything, then process messages serially on a promise + // chain so interleaved chunks cannot drop or reorder responses. + buffer += chunk; + const lines = []; + let newline = buffer.indexOf("\n"); + while (newline >= 0) { + lines.push(buffer.slice(0, newline)); + buffer = buffer.slice(newline + 1); + newline = buffer.indexOf("\n"); + } + for (const rawLine of lines) { + const line = rawLine.trim(); + if (!line) { + continue; } + queue = queue.then(async () => { + let message; + try { + message = JSON.parse(line); + } catch { + writeMessage(errorResponse(null, -32700, "Parse error")); + return; + } + const response = await handleRequest(message, { transport }); + if (response) { + writeMessage(response); + } + }); } }); process.stdin.resume(); @@ -98,44 +129,9 @@ function errorResponse(id, code, message) { }; } -function parseMessages(buffer) { - const messages = []; - let rest = buffer; - while (rest.length > 0) { - const headerEnd = rest.indexOf("\r\n\r\n"); - if (headerEnd >= 0) { - const header = rest.slice(0, headerEnd).toString("utf8"); - const match = header.match(/content-length:\s*(\d+)/iu); - if (!match) { - break; - } - const length = Number(match[1]); - const bodyStart = headerEnd + 4; - const bodyEnd = bodyStart + length; - if (rest.length < bodyEnd) { - break; - } - messages.push(JSON.parse(rest.slice(bodyStart, bodyEnd).toString("utf8"))); - rest = rest.slice(bodyEnd); - continue; - } - - const newline = rest.indexOf("\n"); - if (newline < 0) { - break; - } - const line = rest.slice(0, newline).toString("utf8").trim(); - rest = rest.slice(newline + 1); - if (line) { - messages.push(JSON.parse(line)); - } - } - return { messages, remaining: rest }; -} - function writeMessage(message) { - const body = JSON.stringify(message); - process.stdout.write(`Content-Length: ${Buffer.byteLength(body, "utf8")}\r\n\r\n${body}`); + // MCP stdio transport framing: one JSON-RPC message per line. + process.stdout.write(`${JSON.stringify(message)}\n`); } if (import.meta.url === `file://${process.argv[1]}`) { diff --git a/tests/integrations/runtime-mcp-contract.mjs b/tests/integrations/runtime-mcp-contract.mjs index 8037b75..02d4097 100644 --- a/tests/integrations/runtime-mcp-contract.mjs +++ b/tests/integrations/runtime-mcp-contract.mjs @@ -1,7 +1,17 @@ #!/usr/bin/env node import assert from "node:assert/strict"; -import { handleRequest } from "../../integrations/mcp-server/src/index.mjs"; +import { once } from "node:events"; +import { spawn } from "node:child_process"; +import { fileURLToPath } from "node:url"; +import { + handleRequest, + SUPPORTED_PROTOCOL_VERSIONS, +} from "../../integrations/mcp-server/src/index.mjs"; + +const serverPath = fileURLToPath( + new URL("../../integrations/mcp-server/src/index.mjs", import.meta.url), +); const calls = []; const fakeTransport = { @@ -20,6 +30,29 @@ const fakeTransport = { }, { transport: fakeTransport }); assert.equal(response.result.serverInfo.name, "openphone-runtime"); assert.ok(response.result.capabilities.tools); + assert.equal(response.result.protocolVersion, SUPPORTED_PROTOCOL_VERSIONS[0]); +} + +{ + // A supported client protocolVersion is echoed back. + const response = await handleRequest({ + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2025-06-18" }, + }, { transport: fakeTransport }); + assert.equal(response.result.protocolVersion, "2025-06-18"); +} + +{ + // An unsupported client protocolVersion falls back to the latest we support. + const response = await handleRequest({ + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "1999-01-01" }, + }, { transport: fakeTransport }); + assert.equal(response.result.protocolVersion, SUPPORTED_PROTOCOL_VERSIONS[0]); } { @@ -119,3 +152,72 @@ const fakeTransport = { const response = await handleRequest("not an object", { transport: fakeTransport }); assert.equal(response.error.code, -32600); } + +{ + // Stdio smoke test: a real newline-delimited JSON handshake against the + // server process (initialize -> notifications/initialized -> tools/list). + const server = spawn(process.execPath, [serverPath], { + env: { ...process.env, OPENPHONE_DRY_RUN: "1" }, + stdio: ["pipe", "pipe", "inherit"], + }); + + const responses = []; + let stdoutBuffer = ""; + const waiters = []; + server.stdout.setEncoding("utf8"); + server.stdout.on("data", (chunk) => { + stdoutBuffer += chunk; + let newline = stdoutBuffer.indexOf("\n"); + while (newline >= 0) { + const line = stdoutBuffer.slice(0, newline).trim(); + stdoutBuffer = stdoutBuffer.slice(newline + 1); + if (line) { + responses.push(JSON.parse(line)); + const waiter = waiters.shift(); + if (waiter) { + waiter(); + } + } + newline = stdoutBuffer.indexOf("\n"); + } + }); + + const nextResponse = async () => { + if (responses.length === 0) { + await new Promise((resolve) => waiters.push(resolve)); + } + return responses.shift(); + }; + + const send = (message) => { + server.stdin.write(`${JSON.stringify(message)}\n`); + }; + + try { + send({ + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { + protocolVersion: "2025-06-18", + capabilities: {}, + clientInfo: { name: "contract-test", version: "0.0.0" }, + }, + }); + const initialize = await nextResponse(); + assert.equal(initialize.id, 1); + assert.equal(initialize.result.protocolVersion, "2025-06-18"); + assert.equal(initialize.result.serverInfo.name, "openphone-runtime"); + + send({ jsonrpc: "2.0", method: "notifications/initialized" }); + + send({ jsonrpc: "2.0", id: 2, method: "tools/list", params: {} }); + const toolsList = await nextResponse(); + assert.equal(toolsList.id, 2); + assert.ok(toolsList.result.tools.length > 10); + assert.equal(responses.length, 0); + } finally { + server.kill(); + await once(server, "exit"); + } +}