From 39b73313ec37af6d2950ded7489ac2badf6f860f Mon Sep 17 00:00:00 2001 From: Tom Main Date: Wed, 23 Sep 2026 20:40:26 -0400 Subject: [PATCH 1/2] Send request batches, and add obs-batch The client's sendBatch sends several requests in one RequestBatch message and returns each request's result; a failed request no longer fails its neighbours. obs-batch exposes it with the three OBS execution types, so a serial-frame batch can change two sources on the same frame and pause with Sleep. With OBS_MCP_CONFIRM_LIVE, obs-batch and obs-call-request now ask before sending a request that stops or toggles a stream or recording. obs-preflight reads every input's mute state and tracks in one batch, and obs-capture-window silences a capture in one. The fake OBS answers batches from the same scripts and responders, and records each batched request. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01Vjn8uBEa4mJshaQU3pFuqX --- README.md | 16 +-- src/client.test.ts | 72 ++++++++++- src/client.ts | 139 +++++++++++++++++++-- src/tools/__snapshots__/tool-contract.json | 60 +++++++++ src/tools/batch.test.ts | 88 +++++++++++++ src/tools/confirm.test.ts | 24 ++++ src/tools/confirm.ts | 34 ++++- src/tools/preflight.test.ts | 17 ++- src/tools/preflight.ts | 69 +++++----- src/tools/protocol.ts | 79 +++++++++++- src/tools/workflows.ts | 24 ++-- test/live/obs-live.test.ts | 31 +++++ test/support/fake-obs-server.ts | 90 ++++++++++++- test/support/tool-count.ts | 2 +- 14 files changed, 682 insertions(+), 63 deletions(-) create mode 100644 src/tools/batch.test.ts diff --git a/README.md b/README.md index bafa479..e931e29 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ [![CI](https://github.com/Tom-R-Main/OBS-MCP/actions/workflows/ci.yml/badge.svg)](https://github.com/Tom-R-Main/OBS-MCP/actions/workflows/ci.yml) -OBS MCP gives MCP clients a current, inspectable interface to OBS Studio. It exposes 158 tools for scenes, sources, audio, transitions, filters, recording, streaming, canvases, and the rest of the OBS WebSocket v5 request surface. +OBS MCP gives MCP clients a current, inspectable interface to OBS Studio. It exposes 159 tools for scenes, sources, audio, transitions, filters, recording, streaming, canvases, and the rest of the OBS WebSocket v5 request surface. The server is built against MCP 2026-07-28 and still accepts legacy 2025-era clients. MCP discovery stays available when OBS is closed, and the server reconnects in the background when OBS returns. @@ -106,7 +106,7 @@ The server never prints the password. Connection and protocol diagnostics are wr ## Tool surface -The 158 tools are organized around the OBS protocol rather than a smaller opinionated workflow: +The 159 tools are organized around the OBS protocol rather than a smaller opinionated workflow: - server status, version information, statistics, hotkeys, and studio mode - scenes, groups, sources, filters, and scene items @@ -118,7 +118,7 @@ The 158 tools are organized around the OBS protocol rather than a smaller opinio - `obs-record-clip`, which preflights, records up to 50 seconds with chapter markers, stops, and checks the file's length and audio with ffprobe/ffmpeg when they are installed - `obs-capture-window` (macOS), which points a `screen_capture` input at a window or app by name, silences it, and fits it to the canvas - input and transform changes that return the resulting settings, size, and an optional screenshot, warning when a source renders at 0×0 -- protocol description and the guarded generic request fallback +- protocol description, the guarded generic request fallback, and `obs-batch`, which sends several requests in one message: in order, one per rendered frame (so a change to two sources lands on the same frame), or all at once, with `Sleep` between them Tools that send a single OBS request declare an `outputSchema` generated from the pinned protocol, so clients receive typed `structuredContent` with the documented response fields. Fields are optional and nullable because OBS omits newer fields in older versions and returns undocumented nulls. @@ -143,7 +143,7 @@ Every tool is registered by default. Clients load each registered tool's definit | `outputs` | virtual camera, replay buffer, and generic outputs | | `config` | profiles, scene collections, video settings, persistent data | | `ui` | studio mode, dialogs, projectors | -| `protocol` | `obs-describe-request` and the generic `obs-call-request` | +| `protocol` | `obs-describe-request`, the generic `obs-call-request`, and `obs-batch` | `core` expands to `general`, `scenes`, `scene-items`, `sources`, `inputs`, and `record` (81 tools). `OBS_MCP_TOOLSETS=core OBS_MCP_READ_ONLY=true` gives a 38-tool inspection-only server. The server refuses to start when a group or tool name is unknown. @@ -164,7 +164,7 @@ Two prompts appear as slash commands in clients that show MCP prompts: `record-d ### Confirming live actions -With `OBS_MCP_CONFIRM_LIVE=true`, `obs-stop-stream`, `obs-toggle-stream`, `obs-stop-record`, and `obs-toggle-record` ask before they run. Clients that support elicitation (Claude Code, VS Code, Cursor) show the user a confirmation. Other clients get an error asking the model to get the user's agreement and call again with `confirm: true`. These tools are also marked destructive, so clients that ask before destructive tools already prompt without this setting. +With `OBS_MCP_CONFIRM_LIVE=true`, `obs-stop-stream`, `obs-toggle-stream`, `obs-stop-record`, and `obs-toggle-record` ask before they run, as do `obs-call-request` and `obs-batch` when they carry one of those requests. Clients that support elicitation (Claude Code, VS Code, Cursor) show the user a confirmation. Other clients get an error asking the model to get the user's agreement and call again with `confirm: true`. These tools are also marked destructive, so clients that ask before destructive tools already prompt without this setting. ## Where it fits @@ -184,7 +184,7 @@ Pay particular attention to approvals for tools that: - invoke vendor requests - save files or use the generic `obs-call-request` fallback -`obs-call-request` accepts only request types in the bundled OBS protocol, but its payload is intentionally generic. It is always advertised as destructive and open-world so a client does not silently treat an unfamiliar operation as safe. +`obs-call-request` and `obs-batch` accept only request types in the bundled OBS protocol, but their payloads are intentionally generic. With `OBS_MCP_CONFIRM_LIVE=true`, they ask before sending `StopStream`, `ToggleStream`, `StopRecord`, or `ToggleRecord`. Both are always advertised as destructive and open-world so a client does not silently treat an unfamiliar operation as safe. ## Troubleshooting @@ -215,7 +215,7 @@ Test the file users will actually install with: npm run test:package ``` -That command creates `dist/obs-studio.mcpb`, extracts it into a temporary directory, validates its manifest and source contents, checks all 158 tool definitions, and calls the extracted server through MCP against fake OBS. +That command creates `dist/obs-studio.mcpb`, extracts it into a temporary directory, validates its manifest and source contents, checks all 159 tool definitions, and calls the extracted server through MCP against fake OBS. ### Live OBS tests @@ -225,7 +225,7 @@ The live suite is separate from `npm run check` and CI. By default it is skipped OBS_MCP_LIVE_TEST=1 npm run test:obs-live ``` -The read-only lane checks the OBS version, scenes, active scene collection, streaming state, and recording state through the compiled MCP stdio server. +The read-only lane checks the OBS version, scenes, active scene collection, streaming state, recording state, resources, and read-only request batches through the compiled MCP stdio server. On the machine running OBS, add `OBS_MCP_READ_OBS_CONFIG=1` instead of exporting the password. There is also a narrowly bounded mutation test. Use a disposable scene collection, select it in OBS first, and make sure streaming and recording are stopped. Then run: diff --git a/src/client.test.ts b/src/client.test.ts index ecfbac4..83f4e09 100644 --- a/src/client.test.ts +++ b/src/client.test.ts @@ -7,8 +7,9 @@ import crypto from "node:crypto"; import { once } from "node:events"; import WebSocket from "ws"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import { EventSubscription, OBSWebSocketClient } from "./client.js"; +import { EventSubscription, OBSWebSocketClient, RequestBatchExecutionType } from "./client.js"; import { + FakeOBSRequestError, FakeOBSServer, OBS_OP, type OBSFrame, @@ -235,6 +236,75 @@ describe("OBSWebSocketClient requests", () => { }); }); +describe("OBSWebSocketClient request batches", () => { + it("sends one batch message and returns a result per request, including failures", async () => { + const server = await createServer(); + server.respondWith("GetInputMute", ({ inputName }) => { + if (inputName === "Camera") throw new FakeOBSRequestError(604, "Camera has no audio"); + return { inputMuted: true }; + }); + const client = createClient(server); + + const results = await client.sendBatch([ + { requestType: "GetInputMute", requestData: { inputName: "Mic" } }, + { requestType: "GetInputMute", requestData: { inputName: "Camera" } }, + ], { executionType: RequestBatchExecutionType.Parallel }); + + expect(results).toEqual([ + { requestType: "GetInputMute", ok: true, code: 100, responseData: { inputMuted: true } }, + { requestType: "GetInputMute", ok: false, code: 604, comment: "Camera has no audio", responseData: {} }, + ]); + const batch = server.history().find(({ frame }) => frame.op === OBS_OP.RequestBatch); + expect(batch?.frame.d).toMatchObject({ executionType: 2, haltOnFailure: false }); + expect(server.history().filter(({ frame }) => frame.op === OBS_OP.Request && frame.d.batchRequestId)).toHaveLength(2); + }); + + it("stops at the first failure when asked to", async () => { + const server = await createServer(); + server.respondWith("SetInputMute", () => { + throw new FakeOBSRequestError(600, "No such input"); + }); + server.respondWith("SetInputAudioTracks", () => ({})); + const client = createClient(server); + + const results = await client.sendBatch([ + { requestType: "SetInputMute", requestData: { inputName: "x", inputMuted: true } }, + { requestType: "SetInputAudioTracks", requestData: { inputName: "x", inputAudioTracks: {} } }, + ], { haltOnFailure: true }); + + expect(results.map(({ ok }) => ok)).toEqual([false]); + }); + + it("refuses request types OBS does not advertise before sending anything", async () => { + const server = await createServer(); + const client = createClient(server); + await client.connect(); + const cursor = server.cursor(); + + await expect(client.sendBatch([{ requestType: "MadeUpRequest" }])).rejects.toThrow("MadeUpRequest"); + expect(server.cursor()).toBe(cursor); + }); + + it("extends the timeout by the time the batch sleeps", async () => { + const server = await createServer(); + const client = createClient(server); + + const results = await client.sendBatch([ + { requestType: "Sleep", requestData: { sleepMillis: 150 } }, + ], { timeout: undefined }); + + expect(results).toEqual([{ requestType: "Sleep", ok: true, code: 100, responseData: {} }]); + }); + + it("returns no results without contacting OBS for an empty batch", async () => { + const server = await createServer(); + const client = createClient(server); + + expect(await client.sendBatch([])).toEqual([]); + expect(server.connectionCount).toBe(0); + }); +}); + describe("OBSWebSocketClient events and socket ownership", () => { it("emits both generic and event-specific OBS events", async () => { const server = await createServer(); diff --git a/src/client.ts b/src/client.ts index 926961c..d22c48c 100644 --- a/src/client.ts +++ b/src/client.ts @@ -38,6 +38,34 @@ export enum EventSubscription { All = (1 << 12) - 1, } +/** How OBS runs the requests in a batch (RequestBatchExecutionType). */ +export enum RequestBatchExecutionType { + /** One after another, as fast as possible. */ + SerialRealtime = 0, + /** One per rendered video frame; Sleep counts frames. */ + SerialFrame = 1, + /** All at once; Sleep is not allowed. */ + Parallel = 2, +} + +export type BatchRequest = { requestType: string; requestData?: unknown }; + +export type BatchResult = { + requestType: string; + ok: boolean; + code: number; + comment?: string; + responseData: unknown; +}; + +export type BatchOptions = { + executionType?: RequestBatchExecutionType; + /** Stop at the first failed request; later requests get no result. */ + haltOnFailure?: boolean; + /** Defaults to 10 s plus the batch's own Sleep time. */ + timeout?: number; +}; + type JsonObject = Record; type BaseMessage = { op: number; d: JsonObject }; @@ -140,6 +168,40 @@ function parseRequestResponse(data: JsonObject): RequestResponseData | null { }; } +function parseBatchResponse(data: JsonObject): { requestId: string; results: BatchResult[] } | null { + if (typeof data.requestId !== "string" || !Array.isArray(data.results)) return null; + const results: BatchResult[] = []; + for (const result of data.results) { + if ( + !isObject(result) + || typeof result.requestType !== "string" + || !isObject(result.requestStatus) + || typeof result.requestStatus.result !== "boolean" + || typeof result.requestStatus.code !== "number" + ) return null; + const { comment } = result.requestStatus; + results.push({ + requestType: result.requestType, + ok: result.requestStatus.result, + code: result.requestStatus.code, + ...(typeof comment === "string" ? { comment } : {}), + responseData: result.responseData ?? {}, + }); + } + return { requestId: data.requestId, results }; +} + +/** Wall-clock time a batch spends in Sleep requests, assuming 30 fps for frames. */ +function batchSleepMillis(requests: readonly BatchRequest[]): number { + let total = 0; + for (const { requestType, requestData } of requests) { + if (requestType !== "Sleep" || !isObject(requestData)) continue; + if (typeof requestData.sleepMillis === "number") total += requestData.sleepMillis; + if (typeof requestData.sleepFrames === "number") total += (requestData.sleepFrames / 30) * 1000; + } + return total; +} + function parseEvent(data: JsonObject): EventData | null { if (typeof data.eventType !== "string" || typeof data.eventIntent !== "number") return null; return { @@ -242,6 +304,47 @@ export class OBSWebSocketClient extends EventEmitter { requestData?: unknown, timeout = 10_000, ): Promise { + const socket = await this.readySocket(); + this.assertAdvertised(requestType); + return this.sendTracked(socket, requestType, timeout, (requestId) => ({ + op: OpCode.Request, + d: { + requestType, + requestId, + ...(requestData === undefined ? {} : { requestData }), + }, + })); + } + + /** + * Sends several requests in one message. OBS answers each one; a failed + * request does not reject the batch, so check each result's `ok`. + */ + public async sendBatch(requests: readonly BatchRequest[], options: BatchOptions = {}): Promise { + if (requests.length === 0) return []; + const { + executionType = RequestBatchExecutionType.SerialRealtime, + haltOnFailure = false, + timeout = 10_000 + batchSleepMillis(requests), + } = options; + const socket = await this.readySocket(); + for (const { requestType } of requests) this.assertAdvertised(requestType); + + return this.sendTracked(socket, "RequestBatch", timeout, (requestId) => ({ + op: OpCode.RequestBatch, + d: { + requestId, + haltOnFailure, + executionType, + requests: requests.map(({ requestType, requestData }) => ({ + requestType, + ...(requestData === undefined ? {} : { requestData }), + })), + }, + })); + } + + private async readySocket(): Promise { if (!this.isConnected()) { try { await this.connect(); @@ -257,10 +360,21 @@ export class OBSWebSocketClient extends EventEmitter { if (!socket || socket.readyState !== WebSocket.OPEN) { throw new Error("OBS WebSocket connection was not available after connecting"); } + return socket; + } + + private assertAdvertised(requestType: string): void { if (this.availableRequests && !this.availableRequests.has(requestType)) { throw new Error(`OBS WebSocket does not advertise support for request '${requestType}'`); } + } + private sendTracked( + socket: WebSocket, + requestType: string, + timeout: number, + buildMessage: (requestId: string) => BaseMessage, + ): Promise { return new Promise((resolve, reject) => { const requestId = crypto.randomUUID(); const timeoutId = setTimeout(() => { @@ -279,17 +393,8 @@ export class OBSWebSocketClient extends EventEmitter { }; this.pendingRequests.set(requestId, pending); - const message = { - op: OpCode.Request, - d: { - requestType, - requestId, - ...(requestData === undefined ? {} : { requestData }), - }, - }; - try { - socket.send(JSON.stringify(message), (error) => { + socket.send(JSON.stringify(buildMessage(requestId)), (error) => { if (error) this.rejectPending(requestId, asError(error)); }); } catch (error) { @@ -434,6 +539,12 @@ export class OBSWebSocketClient extends EventEmitter { else logger.error("Ignoring malformed OBS request response"); break; } + case OpCode.RequestBatchResponse: { + const batch = parseBatchResponse(message.d); + if (batch) this.handleBatchResponse(socket, batch); + else logger.error("Ignoring malformed OBS request batch response"); + break; + } case OpCode.Event: { const event = parseEvent(message.d); if (event) this.handleEvent(event); @@ -462,6 +573,14 @@ export class OBSWebSocketClient extends EventEmitter { )); } + private handleBatchResponse(socket: WebSocket, batch: { requestId: string; results: BatchResult[] }): void { + const pending = this.pendingRequests.get(batch.requestId); + if (!pending || pending.socket !== socket) return; + this.pendingRequests.delete(batch.requestId); + clearTimeout(pending.timeout); + pending.resolve(batch.results); + } + private handleEvent(event: EventData): void { this.emit("event", event.eventType, event.eventData); this.emit(event.eventType, event.eventData); diff --git a/src/tools/__snapshots__/tool-contract.json b/src/tools/__snapshots__/tool-contract.json index e027b77..2c2ede9 100644 --- a/src/tools/__snapshots__/tool-contract.json +++ b/src/tools/__snapshots__/tool-contract.json @@ -1,4 +1,64 @@ [ + { + "name": "obs-batch", + "title": "Run OBS Requests in a Batch", + "inputSchema": { + "type": "object", + "properties": { + "requests": { + "minItems": 1, + "maxItems": 100, + "type": "array", + "items": { + "type": "object", + "properties": { + "requestType": { + "type": "string", + "description": "Exact OBS WebSocket request type" + }, + "requestData": { + "description": "Request fields defined by obs-describe-request", + "type": "object", + "propertyNames": { + "type": "string" + }, + "additionalProperties": {} + } + }, + "required": [ + "requestType" + ] + }, + "description": "Requests in the order OBS should run them" + }, + "executionType": { + "default": "serial-realtime", + "description": "How OBS runs the requests", + "type": "string", + "enum": [ + "serial-realtime", + "serial-frame", + "parallel" + ] + }, + "haltOnFailure": { + "default": false, + "description": "Skip the remaining requests after one fails", + "type": "boolean" + } + }, + "required": [ + "requests" + ], + "$schema": "https://json-schema.org/draft/2020-12/schema" + }, + "annotations": { + "readOnlyHint": false, + "destructiveHint": true, + "idempotentHint": false, + "openWorldHint": true + } + }, { "name": "obs-broadcast-custom-event", "title": "Broadcast Custom Event", diff --git a/src/tools/batch.test.ts b/src/tools/batch.test.ts new file mode 100644 index 0000000..f8f25ed --- /dev/null +++ b/src/tools/batch.test.ts @@ -0,0 +1,88 @@ +/* + * Modified for the independent OBS MCP project in August 2026. + * See NOTICE.md and Git history for authorship and change dates. + * SPDX-License-Identifier: GPL-2.0-only + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { FakeOBSRequestError, OBS_OP } from "../../test/support/fake-obs-server.js"; +import { resultText, startMcpHarness, type McpHarness } from "../../test/support/mcp-harness.js"; + +let harness: McpHarness; + +beforeEach(async () => { + vi.spyOn(console, "error").mockImplementation(() => undefined); + harness = await startMcpHarness(); + const { fakeObs } = harness; + fakeObs.respondWith("SetSceneItemEnabled", ({ sceneItemId }) => { + if (sceneItemId === 99) throw new FakeOBSRequestError(600, "No scene item 99"); + return {}; + }); + fakeObs.respondWith("GetCurrentProgramScene", () => ({ sceneName: "Demo" })); +}); + +afterEach(async () => { + await harness.close(); + vi.restoreAllMocks(); +}); + +function batchFrames() { + return harness.fakeObs.history().filter(({ frame }) => frame.op === OBS_OP.RequestBatch); +} + +describe("obs-batch", () => { + it("sends the requests as one batch and reports each result", async () => { + const result = await harness.call("obs-batch", { + executionType: "serial-frame", + requests: [ + { requestType: "SetSceneItemEnabled", requestData: { sceneName: "Demo", sceneItemId: 1, sceneItemEnabled: false } }, + { requestType: "SetSceneItemEnabled", requestData: { sceneName: "Demo", sceneItemId: 2, sceneItemEnabled: true } }, + { requestType: "GetCurrentProgramScene" }, + ], + }); + + expect(result.isError).toBeFalsy(); + expect(batchFrames()).toHaveLength(1); + expect(batchFrames()[0]?.frame.d.executionType).toBe(1); + expect(result.structuredContent).toMatchObject({ succeeded: 3, failed: 0, skipped: 0 }); + expect((result.structuredContent as { results: { responseData: unknown }[] }).results[2]?.responseData) + .toEqual({ sceneName: "Demo" }); + }); + + it("reports failures and the requests skipped after them", async () => { + const result = await harness.call("obs-batch", { + haltOnFailure: true, + requests: [ + { requestType: "SetSceneItemEnabled", requestData: { sceneName: "Demo", sceneItemId: 99, sceneItemEnabled: true } }, + { requestType: "GetCurrentProgramScene" }, + ], + }); + + expect(result.isError).toBe(true); + expect(resultText(result)).toContain("#0 SetSceneItemEnabled failed with code 600: No scene item 99"); + expect(resultText(result)).toContain("1 request(s) skipped"); + expect(result.structuredContent).toMatchObject({ succeeded: 0, failed: 1, skipped: 1 }); + }); + + it("pauses serial batches with Sleep", async () => { + const started = Date.now(); + + const result = await harness.call("obs-batch", { + requests: [{ requestType: "Sleep", requestData: { sleepMillis: 120 } }, { requestType: "GetCurrentProgramScene" }], + }); + + expect(result.isError).toBeFalsy(); + expect(Date.now() - started).toBeGreaterThanOrEqual(110); + }); + + it("refuses unknown requests and Sleep in parallel batches without contacting OBS", async () => { + const unknown = await harness.call("obs-batch", { requests: [{ requestType: "LaunchRocket" }] }); + const parallelSleep = await harness.call("obs-batch", { + executionType: "parallel", + requests: [{ requestType: "Sleep", requestData: { sleepMillis: 10 } }], + }); + + expect(resultText(unknown)).toContain("Unknown OBS WebSocket request: LaunchRocket"); + expect(resultText(parallelSleep)).toContain("Sleep only works in serial"); + expect(batchFrames()).toHaveLength(0); + }); +}); diff --git a/src/tools/confirm.test.ts b/src/tools/confirm.test.ts index 61fdfe4..e20a17e 100644 --- a/src/tools/confirm.test.ts +++ b/src/tools/confirm.test.ts @@ -96,6 +96,30 @@ describe("live action confirmation", () => { expect((await call("obs-stop-record", { confirm: true })).isError).toBeFalsy(); }); + it("asks before generic tools send a live request, and only then", async () => { + const questions: string[] = []; + const { call } = await start({ + confirmLive: true, + onElicit: (message) => { + questions.push(message); + return { action: "decline" }; + }, + }); + harness!.fakeObs.respondWith("GetRecordStatus", () => ({ outputActive: true })); + + const quiet = await call("obs-batch", { requests: [{ requestType: "GetRecordStatus" }] }); + const batch = await call("obs-batch", { + requests: [{ requestType: "GetRecordStatus" }, { requestType: "StopRecord" }], + }); + const single = await call("obs-call-request", { requestType: "StopStream" }); + + expect(quiet.isError).toBeFalsy(); + expect(questions).toEqual(["Stop the recording?", "Stop the live stream?"]); + expect(resultText(batch)).toContain("Cancelled"); + expect(sent("StopRecord")).toBe(false); + expect(sent("StopStream")).toBe(false); + }); + it("confirms through the 2026-07-28 input_required round trip", async () => { vi.spyOn(console, "error").mockImplementation(() => undefined); const fakeObs = await FakeOBSServer.start(); diff --git a/src/tools/confirm.ts b/src/tools/confirm.ts index 9c01517..e1c48e6 100644 --- a/src/tools/confirm.ts +++ b/src/tools/confirm.ts @@ -21,6 +21,30 @@ export const LIVE_ACTIONS: Readonly> = { "obs-toggle-record": "Toggle recording? If it is recording, this ends the take.", }; +/** OBS requests that end a live broadcast or a take, with the question to ask. */ +const LIVE_REQUESTS: Readonly> = { + StopStream: "Stop the live stream?", + ToggleStream: "Toggle the live stream? If it is live, this ends the broadcast.", + StopRecord: "Stop the recording?", + ToggleRecord: "Toggle recording? If it is recording, this ends the take.", +}; + +function questionForRequests(requestTypes: unknown[]): string | undefined { + const questions = [...new Set(requestTypes.map((type) => LIVE_REQUESTS[String(type)]).filter(Boolean))]; + return questions.length > 0 ? questions.join(" ") : undefined; +} + +/** + * Generic tools that are live actions only for some arguments: the question + * to ask, or undefined when these arguments need no confirmation. + */ +export const CONDITIONAL_LIVE_ACTIONS: Readonly) => string | undefined>> = { + "obs-call-request": ({ requestType }) => questionForRequests([requestType]), + "obs-batch": ({ requests }) => questionForRequests( + Array.isArray(requests) ? requests.map((request) => (request as { requestType?: unknown })?.requestType) : [], + ), +}; + const CONFIRM_KEY = "confirm"; const CONFIRM_SCHEMA = z.object({ confirm: z.boolean().describe("Confirm the action") }); @@ -91,7 +115,8 @@ function confirmation( } /** - * Wraps tool registration so the tools in LIVE_ACTIONS ask before running. + * Wraps tool registration so the tools in LIVE_ACTIONS, and the generic + * request tools when they carry a live request, ask before running. * Adds an optional `confirm` argument to their input schema; the original * handler never sees it. */ @@ -101,8 +126,9 @@ export function withLiveConfirmation(server: McpServer): McpServer { if (property === "registerTool") { const registerTool = Reflect.get(target, property, target) as (...args: unknown[]) => unknown; return (name: string, config: { inputSchema?: ZodObject }, callback: (...args: unknown[]) => unknown) => { - const question = LIVE_ACTIONS[name]; - if (!question) return Reflect.apply(registerTool, target, [name, config, callback]); + const fixedQuestion = LIVE_ACTIONS[name]; + const questionFor = CONDITIONAL_LIVE_ACTIONS[name]; + if (!fixedQuestion && !questionFor) return Reflect.apply(registerTool, target, [name, config, callback]); const hadSchema = config.inputSchema !== undefined; const inputSchema = (config.inputSchema ?? z.object({})).extend({ @@ -111,6 +137,8 @@ export function withLiveConfirmation(server: McpServer): McpServer { }); const confirmedCallback = async (args: Record, ctx: RequestContext) => { const { confirm, ...rest } = args; + const question = fixedQuestion ?? questionFor?.(rest); + if (!question) return hadSchema ? callback(rest, ctx) : callback(ctx); const decision = confirmation(target, question, confirm as boolean | undefined, ctx); if (decision !== "proceed") return decision; return hadSchema ? callback(rest, ctx) : callback(ctx); diff --git a/src/tools/preflight.test.ts b/src/tools/preflight.test.ts index fa99777..ee6ed3c 100644 --- a/src/tools/preflight.test.ts +++ b/src/tools/preflight.test.ts @@ -7,7 +7,7 @@ import { mkdtempSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import { FakeOBSRequestError } from "../../test/support/fake-obs-server.js"; +import { FakeOBSRequestError, OBS_OP } from "../../test/support/fake-obs-server.js"; import { resultText, startMcpHarness, type McpHarness } from "../../test/support/mcp-harness.js"; import type { PreflightCheck } from "./preflight.js"; import { healthyObsState, servePreflightState, type FakeObsState } from "../../test/support/fake-obs-state.js"; @@ -117,6 +117,21 @@ describe("obs-preflight", () => { expect(failed.byId("audio")?.status).toBe("fail"); }); + it("reads every input's audio state in one batch", async () => { + state.inputs.push( + { inputName: "Mic", inputKind: "coreaudio_input_capture", muted: false, tracks: { 1: true } }, + { inputName: "Camera", inputKind: "av_capture_input" }, + ); + + const { byId } = await preflight(); + + const batches = harness.fakeObs.history().filter(({ frame }) => frame.op === OBS_OP.RequestBatch); + expect(batches).toHaveLength(1); + expect(batches[0]?.frame.d.requests).toHaveLength(state.inputs.length * 2); + expect(byId("audio")?.message).toContain("Mic (track 1)"); + expect(byId("audio")?.message).not.toContain("Camera"); + }); + it("ignores unmuted inputs whose recorded tracks are disabled", async () => { state.profile["Output/Mode"] = "Advanced"; state.profile["AdvOut/RecEncoder"] = "com.apple.videotoolbox.videoencoder.ave.avc"; diff --git a/src/tools/preflight.ts b/src/tools/preflight.ts index e616a53..3c7df8b 100644 --- a/src/tools/preflight.ts +++ b/src/tools/preflight.ts @@ -6,7 +6,7 @@ import { accessSync, constants, statSync } from "node:fs"; import type { CallToolResult, McpServer } from "@modelcontextprotocol/server"; import { z } from "zod"; -import type { OBSWebSocketClient } from "../client.js"; +import { RequestBatchExecutionType, type OBSWebSocketClient } from "../client.js"; import { READ_ONLY_TOOL } from "./request-tool.js"; export type PreflightStatus = "pass" | "warn" | "fail" | "unknown"; @@ -169,24 +169,33 @@ async function checkAudio(client: OBSWebSocketClient, expectSilent: boolean): Pr request(client, "GetInputList"), recordedTracks(client), ]); + const names = (Array.isArray(inputs) ? inputs : []) + .filter(isObject) + .map(({ inputName }) => inputName) + .filter((name): name is string => typeof name === "string"); + // One round trip for every input's mute state and tracks. Video-only inputs + // fail both requests, which the batch reports without failing the rest. + const results = await client.sendBatch( + names.flatMap((inputName) => [ + { requestType: "GetInputMute", requestData: { inputName } }, + { requestType: "GetInputAudioTracks", requestData: { inputName } }, + ]), + { executionType: RequestBatchExecutionType.Parallel }, + ); const audible: string[] = []; - for (const input of Array.isArray(inputs) ? inputs : []) { - if (!isObject(input) || typeof input.inputName !== "string") continue; - const selector = { inputName: input.inputName }; - let muted: unknown; - try { - ({ inputMuted: muted } = await request(client, "GetInputMute", selector)); - } catch { - continue; // Video-only inputs reject audio requests. - } - if (muted === true) continue; - const { inputAudioTracks } = await request(client, "GetInputAudioTracks", selector); + names.forEach((inputName, index) => { + const mute = results[index * 2]; + const trackResult = results[index * 2 + 1]; + if (!mute?.ok || !isObject(mute.responseData) || mute.responseData.inputMuted === true) return; + const inputAudioTracks = trackResult?.ok && isObject(trackResult.responseData) + ? trackResult.responseData.inputAudioTracks + : undefined; const enabled = isObject(inputAudioTracks) ? tracks.filter((track) => inputAudioTracks[String(track)] === true) : tracks; - if (enabled.length > 0) audible.push(`${input.inputName} (track ${enabled.join(", ")})`); - } + if (enabled.length > 0) audible.push(`${inputName} (track ${enabled.join(", ")})`); + }); if (audible.length === 0) { return { id: "audio", status: "pass", message: "No unmuted input feeds a recorded audio track" }; @@ -207,21 +216,23 @@ async function checkSceneItems(client: OBSWebSocketClient): Promise item.sceneItemEnabled !== false); + // GetSceneItemList includes transforms; look up any it left out in one batch. + const missing = visible.filter((item) => !isObject(item.sceneItemTransform) && typeof item.sceneItemId === "number"); + const looked = await client.sendBatch( + missing.map(({ sceneItemId }) => ({ requestType: "GetSceneItemTransform", requestData: { sceneName, sceneItemId } })), + { executionType: RequestBatchExecutionType.Parallel }, + ); + const transforms = new Map(missing.map((item, index) => { + const result = looked[index]; + return [item, result?.ok && isObject(result.responseData) ? result.responseData.sceneItemTransform : undefined]; + })); + const blank = visible + .filter((item) => { + const transform = isObject(item.sceneItemTransform) ? item.sceneItemTransform : transforms.get(item); + return isObject(transform) && (transform.sourceWidth === 0 || transform.sourceHeight === 0); + }) + .map((item) => String(item.sourceName ?? item.sceneItemId)); if (items.length === 0) { return { id: "scene-items", status: "warn", message: `Program scene ${sceneName} is empty` }; diff --git a/src/tools/protocol.ts b/src/tools/protocol.ts index 3226b4b..70402c9 100644 --- a/src/tools/protocol.ts +++ b/src/tools/protocol.ts @@ -5,7 +5,7 @@ */ import type { McpServer } from "@modelcontextprotocol/server"; import { z } from "zod"; -import type { OBSWebSocketClient } from "../client.js"; +import { RequestBatchExecutionType, type OBSWebSocketClient } from "../client.js"; import { getObsProtocolRequest, OBS_PROTOCOL_REQUESTS } from "../obs-protocol.js"; import { READ_ONLY_TOOL } from "./request-tool.js"; @@ -16,6 +16,14 @@ const UNRESTRICTED_REQUEST_TOOL = { openWorldHint: true, } as const; +const MAX_BATCH_REQUESTS = 100; + +const EXECUTION_TYPES = { + "serial-realtime": RequestBatchExecutionType.SerialRealtime, + "serial-frame": RequestBatchExecutionType.SerialFrame, + parallel: RequestBatchExecutionType.Parallel, +} as const; + export function initialize(server: McpServer, client: OBSWebSocketClient): void { server.registerTool( "obs-describe-request", @@ -95,4 +103,73 @@ export function initialize(server: McpServer, client: OBSWebSocketClient): void } }, ); + + server.registerTool( + "obs-batch", + { + title: "Run OBS Requests in a Batch", + description: "Send several requests from the pinned OBS WebSocket protocol in one message. " + + "serial-realtime runs them in order as fast as possible; serial-frame runs one per rendered frame, " + + "so changes such as hiding one source and showing another land on the same frame; parallel runs " + + "them together. Sleep ({sleepMillis} or {sleepFrames}) pauses serial batches. Each request " + + "succeeds or fails on its own unless haltOnFailure is set", + inputSchema: z.object({ + requests: z.array(z.object({ + requestType: z.string().describe("Exact OBS WebSocket request type"), + requestData: z.record(z.string(), z.unknown()).optional().describe("Request fields defined by obs-describe-request"), + })).min(1).max(MAX_BATCH_REQUESTS).describe("Requests in the order OBS should run them"), + executionType: z.enum(["serial-realtime", "serial-frame", "parallel"]).default("serial-realtime") + .describe("How OBS runs the requests"), + haltOnFailure: z.boolean().default(false).describe("Skip the remaining requests after one fails"), + }), + annotations: UNRESTRICTED_REQUEST_TOOL, + }, + async ({ requests, executionType, haltOnFailure }) => { + const unknown = requests.filter(({ requestType }) => !getObsProtocolRequest(requestType)); + if (unknown.length > 0) { + return { + content: [{ type: "text", text: `Unknown OBS WebSocket request: ${unknown.map(({ requestType }) => requestType).join(", ")}` }], + isError: true, + }; + } + if (executionType === "parallel" && requests.some(({ requestType }) => requestType === "Sleep")) { + return { + content: [{ type: "text", text: "Sleep only works in serial-realtime and serial-frame batches" }], + isError: true, + }; + } + + try { + const batch = await client.sendBatch(requests, { executionType: EXECUTION_TYPES[executionType], haltOnFailure }); + const results = batch.map((result, index) => ({ index, ...result })); + const failed = results.filter(({ ok }) => !ok); + const structuredContent = { + results, + succeeded: results.length - failed.length, + failed: failed.length, + skipped: requests.length - results.length, + }; + const summary = [ + `${structuredContent.succeeded} of ${requests.length} request(s) succeeded`, + ...failed.map(({ index, requestType, code, comment }) => ( + `#${index} ${requestType} failed with code ${code}${comment ? `: ${comment}` : ""}` + )), + ...(structuredContent.skipped > 0 ? [`${structuredContent.skipped} request(s) skipped after the failure`] : []), + ]; + return { + content: [{ type: "text", text: `${summary.join("\n")}\n${JSON.stringify(results, null, 2)}` }], + structuredContent, + ...(failed.length > 0 ? { isError: true } : {}), + }; + } catch (error) { + return { + content: [{ + type: "text", + text: `OBS request batch failed: ${error instanceof Error ? error.message : String(error)}`, + }], + isError: true, + }; + } + }, + ); } diff --git a/src/tools/workflows.ts b/src/tools/workflows.ts index a603bbb..e92f350 100644 --- a/src/tools/workflows.ts +++ b/src/tools/workflows.ts @@ -7,7 +7,7 @@ import { execFile } from "node:child_process"; import { promisify } from "node:util"; import type { CallToolResult, McpServer } from "@modelcontextprotocol/server"; import { z } from "zod"; -import type { OBSWebSocketClient } from "../client.js"; +import { RequestBatchExecutionType, type OBSWebSocketClient } from "../client.js"; import { appendInputScreenshot, inputStateResult, readInputState } from "./after-change.js"; import { RECORD_OUTPUT, startOutputAndConfirm } from "./output-start.js"; import { runPreflight } from "./preflight.js"; @@ -233,17 +233,25 @@ async function captureWindow( const notes: string[] = [`Capturing ${property} ${match.itemName}`]; if (args.silent) { // screen_capture carries the application's audio on macOS 13+. - await client.sendRequest("SetInputMute", { inputName: args.inputName, inputMuted: true }); - await client.sendRequest("SetInputAudioTracks", { inputName: args.inputName, inputAudioTracks: ALL_TRACKS_OFF }); + const silenced = await client.sendBatch([ + { requestType: "SetInputMute", requestData: { inputName: args.inputName, inputMuted: true } }, + { requestType: "SetInputAudioTracks", requestData: { inputName: args.inputName, inputAudioTracks: ALL_TRACKS_OFF } }, + ], { haltOnFailure: true }); + const failure = silenced.find(({ ok }) => !ok) ?? (silenced.length < 2 ? silenced[0] : undefined); + if (failure) throw new Error(`${failure.requestType} failed with code ${failure.code}${failure.comment ? `: ${failure.comment}` : ""}`); notes.push("Muted the capture and removed it from every audio track"); } if (args.fit) { - const video = await client.sendRequest("GetVideoSettings") as JsonObject; - const { sceneItemId } = await client.sendRequest("GetSceneItemId", { - sceneName: args.sceneName, - sourceName: args.inputName, - }) as { sceneItemId: number }; + const [videoResult, itemResult] = await client.sendBatch([ + { requestType: "GetVideoSettings" }, + { requestType: "GetSceneItemId", requestData: { sceneName: args.sceneName, sourceName: args.inputName } }, + ], { executionType: RequestBatchExecutionType.Parallel }); + for (const result of [videoResult, itemResult]) { + if (!result?.ok) throw new Error(`${result?.requestType ?? "Request"} failed${result?.comment ? `: ${result.comment}` : ""}`); + } + const video = videoResult!.responseData as JsonObject; + const { sceneItemId } = itemResult!.responseData as { sceneItemId: number }; await client.sendRequest("SetSceneItemTransform", { sceneName: args.sceneName, sceneItemId, diff --git a/test/live/obs-live.test.ts b/test/live/obs-live.test.ts index 4494b5c..a90b691 100644 --- a/test/live/obs-live.test.ts +++ b/test/live/obs-live.test.ts @@ -66,6 +66,9 @@ describe.skipIf(!enabled)("live OBS through compiled MCP stdio", () => { ...(process.env.OBS_WEBSOCKET_PASSWORD ? { OBS_WEBSOCKET_PASSWORD: process.env.OBS_WEBSOCKET_PASSWORD } : {}), + ...(process.env.OBS_MCP_READ_OBS_CONFIG + ? { OBS_MCP_READ_OBS_CONFIG: process.env.OBS_MCP_READ_OBS_CONFIG } + : {}), }, stderr: "pipe", }); @@ -135,6 +138,34 @@ describe.skipIf(!enabled)("live OBS through compiled MCP stdio", () => { expect(mismatches).toEqual([]); }, 60_000); + it("runs read-only request batches, including Sleep, against real OBS", async () => { + const batch = async (args: JsonObject) => await client!.callTool({ name: "obs-batch", arguments: args }) as CallToolResult; + + // A missing input fails on its own without failing the rest. + const parallel = await batch({ + executionType: "parallel", + requests: [ + { requestType: "GetVersion" }, + { requestType: "GetStats" }, + { requestType: "GetInputMute", requestData: { inputName: "obs-mcp-no-such-input" } }, + ], + }); + expect(parallel.structuredContent).toMatchObject({ succeeded: 2, failed: 1, skipped: 0 }); + + const started = Date.now(); + const framed = await batch({ + executionType: "serial-frame", + requests: [ + { requestType: "GetCurrentProgramScene" }, + { requestType: "Sleep", requestData: { sleepFrames: 6 } }, + { requestType: "GetRecordStatus" }, + ], + }); + expect(framed.isError).toBeFalsy(); + expect(framed.structuredContent).toMatchObject({ succeeded: 3, failed: 0 }); + expect(Date.now() - started).toBeGreaterThanOrEqual(60); + }); + it("reads status, scene, and scene item resources from real OBS", async () => { if (!client) throw new Error("Live MCP client is not connected"); const read = async (uri: string) => { diff --git a/test/support/fake-obs-server.ts b/test/support/fake-obs-server.ts index 8629ca4..46d5668 100644 --- a/test/support/fake-obs-server.ts +++ b/test/support/fake-obs-server.ts @@ -15,8 +15,13 @@ export const OBS_OP = { Event: 5, Request: 6, RequestResponse: 7, + RequestBatch: 8, + RequestBatchResponse: 9, } as const; +/** Milliseconds per frame when a fake batch sleeps for frames. */ +const FAKE_FRAME_MS = 1000 / 60; + type JsonObject = Record; export type OBSFrame = { @@ -136,7 +141,7 @@ export class FakeOBSServer { this.obsStudioVersion = options.obsStudioVersion ?? "32.2.2"; this.obsWebSocketVersion = options.obsWebSocketVersion ?? "5.7.0"; this.platform = options.platform ?? "macos"; - this.availableRequests = Array.from(new Set(["GetVersion", ...(options.availableRequests ?? [])])); + this.availableRequests = Array.from(new Set(["GetVersion", "Sleep", ...(options.availableRequests ?? [])])); const address = server.address() as AddressInfo; this.url = `ws://127.0.0.1:${address.port}`; @@ -343,6 +348,11 @@ export class FakeOBSServer { return; } + if (frame.op === OBS_OP.RequestBatch) { + void this.handleBatch(record); + return; + } + if (!this.isRequest(record)) return; const action = this.scripts.get(record.frame.d.requestType)?.shift(); if (action) { @@ -369,6 +379,84 @@ export class FakeOBSServer { } } + /** + * Answers a request batch from the same scripts and responders as single + * requests. Each request in it is recorded in history() as its own Request + * frame carrying batchRequestId, so request assertions see batched requests. + */ + private async handleBatch({ frame, socket, connectionId }: InternalRecordedFrame): Promise { + const batchRequestId = String(frame.d.requestId); + const executionType = typeof frame.d.executionType === "number" ? frame.d.executionType : 0; + const requests = Array.isArray(frame.d.requests) ? frame.d.requests.filter(isObject) : []; + const results: JsonObject[] = []; + + for (const [index, request] of requests.entries()) { + const requestType = String(request.requestType); + const requestData = isObject(request.requestData) ? request.requestData : {}; + const subRecord: InternalRecordedFrame = { + cursor: this.nextCursor++, + connectionId, + socket, + frame: { + op: OBS_OP.Request, + d: { requestType, requestId: `${batchRequestId}#${index}`, requestData, batchRequestId }, + }, + }; + this.frames.push(subRecord); + this.events.emit("frame", subRecord); + + const answer = await this.answerBatchedRequest(requestType, requestData, executionType); + if (answer === "disconnect") { + socket.close(1011, "scripted disconnect"); + return; + } + results.push({ requestType, requestStatus: answer.status, responseData: answer.data }); + if (!answer.status.result && frame.d.haltOnFailure === true) break; + } + + if (socket.readyState !== WebSocket.OPEN) return; + socket.send(JSON.stringify({ op: OBS_OP.RequestBatchResponse, d: { requestId: batchRequestId, results } })); + } + + private async answerBatchedRequest( + requestType: string, + requestData: JsonObject, + executionType: number, + ): Promise<{ status: ResponseStatus; data: JsonObject } | "disconnect"> { + if (requestType === "Sleep") { + if (executionType === 2) { + return { status: { result: false, code: 206, comment: "Sleep is not available in parallel batches" }, data: {} }; + } + const millis = typeof requestData.sleepMillis === "number" + ? requestData.sleepMillis + : Number(requestData.sleepFrames ?? 0) * FAKE_FRAME_MS; + await new Promise((resolve) => setTimeout(resolve, millis)); + return { status: { result: true, code: 100 }, data: {} }; + } + + const action = this.scripts.get(requestType)?.shift(); + if (action?.kind === "disconnect") return "disconnect"; + if (action) return { status: action.status, data: action.data }; + + const responder = this.responders.get(requestType); + if (responder) { + try { + return { status: { result: true, code: 100 }, data: responder(requestData) }; + } catch (error) { + return { + status: error instanceof FakeOBSRequestError + ? { result: false, code: error.code, comment: error.message } + : { result: false, code: 500, comment: String(error) }, + data: {}, + }; + } + } + if (requestType === "GetVersion" && this.autoGetVersion) { + return { status: { result: true, code: 100 }, data: this.versionResponse() }; + } + return { status: { result: false, code: 204, comment: `Fake OBS has no answer for ${requestType}` }, data: {} }; + } + private queueAction(requestType: string, action: ScriptedAction): void { const actions = this.scripts.get(requestType) ?? []; actions.push(action); diff --git a/test/support/tool-count.ts b/test/support/tool-count.ts index 254efbe..2f2ef93 100644 --- a/test/support/tool-count.ts +++ b/test/support/tool-count.ts @@ -5,4 +5,4 @@ */ /** Number of tools the server registers with no tool filtering configured. */ -export const TOOL_COUNT = 158; +export const TOOL_COUNT = 159; From b0ed6320371ea4f271326720bafdf13a784de07b Mon Sep 17 00:00:00 2001 From: Tom Main Date: Wed, 23 Sep 2026 21:30:29 -0400 Subject: [PATCH 2/2] Stop using parallel request batches In obs-websocket 5.7, a parallel batch holds a thread-pool thread while it waits for its requests on the same pool (RequestBatchHandler.cpp), so a few concurrent parallel batches deadlock the WebSocket server until OBS restarts. Its results are also collected in completion order, so they came back matched to the wrong requests: live, GetInputSettings for a color source was reported as failing with "does not support audio". Every internal batch now runs serial-realtime, still one round trip. obs-batch no longer offers parallel. Each batched request carries its index as its ID and results are matched by ID. The live test now checks each result's position. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01Vjn8uBEa4mJshaQU3pFuqX --- README.md | 2 +- src/client.test.ts | 7 +++-- src/client.ts | 34 +++++++++++++++++----- src/tools/__snapshots__/tool-contract.json | 3 +- src/tools/batch.test.ts | 9 +++--- src/tools/preflight.ts | 14 ++++----- src/tools/protocol.ts | 13 ++------- src/tools/workflows.ts | 4 +-- test/live/obs-live.test.ts | 7 +++-- test/support/fake-obs-server.ts | 9 +++++- 10 files changed, 58 insertions(+), 44 deletions(-) diff --git a/README.md b/README.md index e931e29..a33bcfd 100644 --- a/README.md +++ b/README.md @@ -118,7 +118,7 @@ The 159 tools are organized around the OBS protocol rather than a smaller opinio - `obs-record-clip`, which preflights, records up to 50 seconds with chapter markers, stops, and checks the file's length and audio with ffprobe/ffmpeg when they are installed - `obs-capture-window` (macOS), which points a `screen_capture` input at a window or app by name, silences it, and fits it to the canvas - input and transform changes that return the resulting settings, size, and an optional screenshot, warning when a source renders at 0×0 -- protocol description, the guarded generic request fallback, and `obs-batch`, which sends several requests in one message: in order, one per rendered frame (so a change to two sources lands on the same frame), or all at once, with `Sleep` between them +- protocol description, the guarded generic request fallback, and `obs-batch`, which sends several requests in one message, in order or one per rendered frame (so a change to two sources lands on the same frame), with `Sleep` between them. OBS's parallel mode is not offered: in obs-websocket 5.7, concurrent parallel batches deadlock its WebSocket server until OBS restarts Tools that send a single OBS request declare an `outputSchema` generated from the pinned protocol, so clients receive typed `structuredContent` with the documented response fields. Fields are optional and nullable because OBS omits newer fields in older versions and returns undocumented nulls. diff --git a/src/client.test.ts b/src/client.test.ts index 83f4e09..859f352 100644 --- a/src/client.test.ts +++ b/src/client.test.ts @@ -7,7 +7,7 @@ import crypto from "node:crypto"; import { once } from "node:events"; import WebSocket from "ws"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import { EventSubscription, OBSWebSocketClient, RequestBatchExecutionType } from "./client.js"; +import { EventSubscription, OBSWebSocketClient } from "./client.js"; import { FakeOBSRequestError, FakeOBSServer, @@ -248,14 +248,15 @@ describe("OBSWebSocketClient request batches", () => { const results = await client.sendBatch([ { requestType: "GetInputMute", requestData: { inputName: "Mic" } }, { requestType: "GetInputMute", requestData: { inputName: "Camera" } }, - ], { executionType: RequestBatchExecutionType.Parallel }); + ]); expect(results).toEqual([ { requestType: "GetInputMute", ok: true, code: 100, responseData: { inputMuted: true } }, { requestType: "GetInputMute", ok: false, code: 604, comment: "Camera has no audio", responseData: {} }, ]); const batch = server.history().find(({ frame }) => frame.op === OBS_OP.RequestBatch); - expect(batch?.frame.d).toMatchObject({ executionType: 2, haltOnFailure: false }); + expect(batch?.frame.d).toMatchObject({ executionType: 0, haltOnFailure: false }); + expect((batch?.frame.d.requests as { requestId: string }[]).map(({ requestId }) => requestId)).toEqual(["0", "1"]); expect(server.history().filter(({ frame }) => frame.op === OBS_OP.Request && frame.d.batchRequestId)).toHaveLength(2); }); diff --git a/src/client.ts b/src/client.ts index d22c48c..c611b2a 100644 --- a/src/client.ts +++ b/src/client.ts @@ -38,14 +38,18 @@ export enum EventSubscription { All = (1 << 12) - 1, } -/** How OBS runs the requests in a batch (RequestBatchExecutionType). */ +/** + * How OBS runs the requests in a batch (RequestBatchExecutionType). Parallel + * (2) is left out on purpose: in obs-websocket 5.7 each parallel batch holds a + * thread-pool thread while waiting for its requests on the same pool, so a few + * at once deadlock the WebSocket server until OBS restarts, and its results + * come back in completion order with the wrong request labels. + */ export enum RequestBatchExecutionType { /** One after another, as fast as possible. */ SerialRealtime = 0, /** One per rendered video frame; Sleep counts frames. */ SerialFrame = 1, - /** All at once; Sleep is not allowed. */ - Parallel = 2, } export type BatchRequest = { requestType: string; requestData?: unknown }; @@ -168,9 +172,11 @@ function parseRequestResponse(data: JsonObject): RequestResponseData | null { }; } -function parseBatchResponse(data: JsonObject): { requestId: string; results: BatchResult[] } | null { +type RawBatchResult = BatchResult & { requestId?: string }; + +function parseBatchResponse(data: JsonObject): { requestId: string; results: RawBatchResult[] } | null { if (typeof data.requestId !== "string" || !Array.isArray(data.results)) return null; - const results: BatchResult[] = []; + const results: RawBatchResult[] = []; for (const result of data.results) { if ( !isObject(result) @@ -181,6 +187,7 @@ function parseBatchResponse(data: JsonObject): { requestId: string; results: Bat ) return null; const { comment } = result.requestStatus; results.push({ + ...(typeof result.requestId === "string" ? { requestId: result.requestId } : {}), requestType: result.requestType, ok: result.requestStatus.result, code: result.requestStatus.code, @@ -191,6 +198,14 @@ function parseBatchResponse(data: JsonObject): { requestId: string; results: Bat return { requestId: data.requestId, results }; } +/** Orders batch results by the index each request carried as its ID, dropping the IDs. */ +function inRequestOrder(results: RawBatchResult[]): BatchResult[] { + const indexed = results.map((result, position) => ({ result, index: result.requestId === undefined ? position : Number(result.requestId) })); + return indexed + .sort((a, b) => a.index - b.index) + .map(({ result: { requestId: _requestId, ...result } }) => result); +} + /** Wall-clock time a batch spends in Sleep requests, assuming 30 fps for frames. */ function batchSleepMillis(requests: readonly BatchRequest[]): number { let total = 0; @@ -330,18 +345,21 @@ export class OBSWebSocketClient extends EventEmitter { const socket = await this.readySocket(); for (const { requestType } of requests) this.assertAdvertised(requestType); - return this.sendTracked(socket, "RequestBatch", timeout, (requestId) => ({ + const results = await this.sendTracked(socket, "RequestBatch", timeout, (requestId) => ({ op: OpCode.RequestBatch, d: { requestId, haltOnFailure, executionType, - requests: requests.map(({ requestType, requestData }) => ({ + // Each request carries its index as its ID, so results are matched by ID, not position. + requests: requests.map(({ requestType, requestData }, index) => ({ requestType, + requestId: String(index), ...(requestData === undefined ? {} : { requestData }), })), }, })); + return inRequestOrder(results); } private async readySocket(): Promise { @@ -573,7 +591,7 @@ export class OBSWebSocketClient extends EventEmitter { )); } - private handleBatchResponse(socket: WebSocket, batch: { requestId: string; results: BatchResult[] }): void { + private handleBatchResponse(socket: WebSocket, batch: { requestId: string; results: RawBatchResult[] }): void { const pending = this.pendingRequests.get(batch.requestId); if (!pending || pending.socket !== socket) return; this.pendingRequests.delete(batch.requestId); diff --git a/src/tools/__snapshots__/tool-contract.json b/src/tools/__snapshots__/tool-contract.json index 2c2ede9..84285aa 100644 --- a/src/tools/__snapshots__/tool-contract.json +++ b/src/tools/__snapshots__/tool-contract.json @@ -37,8 +37,7 @@ "type": "string", "enum": [ "serial-realtime", - "serial-frame", - "parallel" + "serial-frame" ] }, "haltOnFailure": { diff --git a/src/tools/batch.test.ts b/src/tools/batch.test.ts index f8f25ed..fc8d38e 100644 --- a/src/tools/batch.test.ts +++ b/src/tools/batch.test.ts @@ -74,15 +74,16 @@ describe("obs-batch", () => { expect(Date.now() - started).toBeGreaterThanOrEqual(110); }); - it("refuses unknown requests and Sleep in parallel batches without contacting OBS", async () => { + it("refuses unknown requests and the parallel mode without contacting OBS", async () => { const unknown = await harness.call("obs-batch", { requests: [{ requestType: "LaunchRocket" }] }); - const parallelSleep = await harness.call("obs-batch", { + // obs-websocket 5.7 deadlocks on concurrent parallel batches, so the mode is not offered. + const parallel = await harness.call("obs-batch", { executionType: "parallel", - requests: [{ requestType: "Sleep", requestData: { sleepMillis: 10 } }], + requests: [{ requestType: "GetCurrentProgramScene" }], }); expect(resultText(unknown)).toContain("Unknown OBS WebSocket request: LaunchRocket"); - expect(resultText(parallelSleep)).toContain("Sleep only works in serial"); + expect(parallel.isError).toBe(true); expect(batchFrames()).toHaveLength(0); }); }); diff --git a/src/tools/preflight.ts b/src/tools/preflight.ts index 3c7df8b..ca675a6 100644 --- a/src/tools/preflight.ts +++ b/src/tools/preflight.ts @@ -6,7 +6,7 @@ import { accessSync, constants, statSync } from "node:fs"; import type { CallToolResult, McpServer } from "@modelcontextprotocol/server"; import { z } from "zod"; -import { RequestBatchExecutionType, type OBSWebSocketClient } from "../client.js"; +import type { OBSWebSocketClient } from "../client.js"; import { READ_ONLY_TOOL } from "./request-tool.js"; export type PreflightStatus = "pass" | "warn" | "fail" | "unknown"; @@ -175,13 +175,10 @@ async function checkAudio(client: OBSWebSocketClient, expectSilent: boolean): Pr .filter((name): name is string => typeof name === "string"); // One round trip for every input's mute state and tracks. Video-only inputs // fail both requests, which the batch reports without failing the rest. - const results = await client.sendBatch( - names.flatMap((inputName) => [ - { requestType: "GetInputMute", requestData: { inputName } }, - { requestType: "GetInputAudioTracks", requestData: { inputName } }, - ]), - { executionType: RequestBatchExecutionType.Parallel }, - ); + const results = await client.sendBatch(names.flatMap((inputName) => [ + { requestType: "GetInputMute", requestData: { inputName } }, + { requestType: "GetInputAudioTracks", requestData: { inputName } }, + ])); const audible: string[] = []; names.forEach((inputName, index) => { @@ -221,7 +218,6 @@ async function checkSceneItems(client: OBSWebSocketClient): Promise !isObject(item.sceneItemTransform) && typeof item.sceneItemId === "number"); const looked = await client.sendBatch( missing.map(({ sceneItemId }) => ({ requestType: "GetSceneItemTransform", requestData: { sceneName, sceneItemId } })), - { executionType: RequestBatchExecutionType.Parallel }, ); const transforms = new Map(missing.map((item, index) => { const result = looked[index]; diff --git a/src/tools/protocol.ts b/src/tools/protocol.ts index 70402c9..de6b3da 100644 --- a/src/tools/protocol.ts +++ b/src/tools/protocol.ts @@ -21,7 +21,6 @@ const MAX_BATCH_REQUESTS = 100; const EXECUTION_TYPES = { "serial-realtime": RequestBatchExecutionType.SerialRealtime, "serial-frame": RequestBatchExecutionType.SerialFrame, - parallel: RequestBatchExecutionType.Parallel, } as const; export function initialize(server: McpServer, client: OBSWebSocketClient): void { @@ -110,15 +109,14 @@ export function initialize(server: McpServer, client: OBSWebSocketClient): void title: "Run OBS Requests in a Batch", description: "Send several requests from the pinned OBS WebSocket protocol in one message. " + "serial-realtime runs them in order as fast as possible; serial-frame runs one per rendered frame, " - + "so changes such as hiding one source and showing another land on the same frame; parallel runs " - + "them together. Sleep ({sleepMillis} or {sleepFrames}) pauses serial batches. Each request " + + "so changes such as hiding one source and showing another land on the same frame. Sleep ({sleepMillis} or {sleepFrames}) pauses serial batches. Each request " + "succeeds or fails on its own unless haltOnFailure is set", inputSchema: z.object({ requests: z.array(z.object({ requestType: z.string().describe("Exact OBS WebSocket request type"), requestData: z.record(z.string(), z.unknown()).optional().describe("Request fields defined by obs-describe-request"), })).min(1).max(MAX_BATCH_REQUESTS).describe("Requests in the order OBS should run them"), - executionType: z.enum(["serial-realtime", "serial-frame", "parallel"]).default("serial-realtime") + executionType: z.enum(["serial-realtime", "serial-frame"]).default("serial-realtime") .describe("How OBS runs the requests"), haltOnFailure: z.boolean().default(false).describe("Skip the remaining requests after one fails"), }), @@ -132,13 +130,6 @@ export function initialize(server: McpServer, client: OBSWebSocketClient): void isError: true, }; } - if (executionType === "parallel" && requests.some(({ requestType }) => requestType === "Sleep")) { - return { - content: [{ type: "text", text: "Sleep only works in serial-realtime and serial-frame batches" }], - isError: true, - }; - } - try { const batch = await client.sendBatch(requests, { executionType: EXECUTION_TYPES[executionType], haltOnFailure }); const results = batch.map((result, index) => ({ index, ...result })); diff --git a/src/tools/workflows.ts b/src/tools/workflows.ts index e92f350..09c6e1a 100644 --- a/src/tools/workflows.ts +++ b/src/tools/workflows.ts @@ -7,7 +7,7 @@ import { execFile } from "node:child_process"; import { promisify } from "node:util"; import type { CallToolResult, McpServer } from "@modelcontextprotocol/server"; import { z } from "zod"; -import { RequestBatchExecutionType, type OBSWebSocketClient } from "../client.js"; +import type { OBSWebSocketClient } from "../client.js"; import { appendInputScreenshot, inputStateResult, readInputState } from "./after-change.js"; import { RECORD_OUTPUT, startOutputAndConfirm } from "./output-start.js"; import { runPreflight } from "./preflight.js"; @@ -246,7 +246,7 @@ async function captureWindow( const [videoResult, itemResult] = await client.sendBatch([ { requestType: "GetVideoSettings" }, { requestType: "GetSceneItemId", requestData: { sceneName: args.sceneName, sourceName: args.inputName } }, - ], { executionType: RequestBatchExecutionType.Parallel }); + ]); for (const result of [videoResult, itemResult]) { if (!result?.ok) throw new Error(`${result?.requestType ?? "Request"} failed${result?.comment ? `: ${result.comment}` : ""}`); } diff --git a/test/live/obs-live.test.ts b/test/live/obs-live.test.ts index a90b691..db732bf 100644 --- a/test/live/obs-live.test.ts +++ b/test/live/obs-live.test.ts @@ -142,15 +142,16 @@ describe.skipIf(!enabled)("live OBS through compiled MCP stdio", () => { const batch = async (args: JsonObject) => await client!.callTool({ name: "obs-batch", arguments: args }) as CallToolResult; // A missing input fails on its own without failing the rest. - const parallel = await batch({ - executionType: "parallel", + const realtime = await batch({ requests: [ { requestType: "GetVersion" }, { requestType: "GetStats" }, { requestType: "GetInputMute", requestData: { inputName: "obs-mcp-no-such-input" } }, ], }); - expect(parallel.structuredContent).toMatchObject({ succeeded: 2, failed: 1, skipped: 0 }); + expect(realtime.structuredContent).toMatchObject({ succeeded: 2, failed: 1, skipped: 0 }); + const results = (realtime.structuredContent as { results: { requestType: string; ok: boolean }[] }).results; + expect(results.map(({ requestType, ok }) => `${requestType}:${ok}`)).toEqual(["GetVersion:true", "GetStats:true", "GetInputMute:false"]); const started = Date.now(); const framed = await batch({ diff --git a/test/support/fake-obs-server.ts b/test/support/fake-obs-server.ts index 46d5668..f50bb33 100644 --- a/test/support/fake-obs-server.ts +++ b/test/support/fake-obs-server.ts @@ -410,11 +410,18 @@ export class FakeOBSServer { socket.close(1011, "scripted disconnect"); return; } - results.push({ requestType, requestStatus: answer.status, responseData: answer.data }); + results.push({ + requestType, + ...(typeof request.requestId === "string" ? { requestId: request.requestId } : {}), + requestStatus: answer.status, + responseData: answer.data, + }); if (!answer.status.result && frame.d.haltOnFailure === true) break; } if (socket.readyState !== WebSocket.OPEN) return; + // Real OBS does not return parallel results in request order; neither does the fake. + if (executionType === 2) results.reverse(); socket.send(JSON.stringify({ op: OBS_OP.RequestBatchResponse, d: { requestId: batchRequestId, results } })); }