From 8e28bca775b66301a9b8f14916f84eebdfbd0e17 Mon Sep 17 00:00:00 2001 From: DrSmoothl <1787882683@qq.com> Date: Fri, 23 Oct 2026 21:19:37 +0800 Subject: [PATCH] feat(mcp): add compatible protocol workflows and execution services --- AGENTS.md | 3 +- packages/mcp/AGENT_FEATURES.md | 250 ++++- packages/mcp/apps/message-preview.ts | 9 + packages/mcp/apps/preview-view.ts | 196 ++++ packages/mcp/apps/tsconfig.json | 9 + packages/mcp/package.json | 13 +- packages/mcp/scripts/build-message-app.mjs | 48 + packages/mcp/src/client.ts | 20 +- packages/mcp/src/elicitation.ts | 344 +++++++ packages/mcp/src/message-app.ts | 157 +++ packages/mcp/src/results.ts | 2 +- packages/mcp/src/server-factory.ts | 373 ++++++- packages/mcp/src/server.ts | 103 +- packages/mcp/src/shutdown.ts | 89 ++ packages/mcp/src/subscriptions.ts | 722 +++++++++++++ packages/mcp/src/tasks.ts | 956 ++++++++++++++++++ packages/mcp/src/tools.ts | 59 +- packages/mcp/src/tracing.ts | 110 ++ packages/mcp/tests/elicitation.test.ts | 585 +++++++++++ packages/mcp/tests/message-app.test.ts | 474 +++++++++ packages/mcp/tests/protocol-versions.test.ts | 637 ++++++++++++ packages/mcp/tests/server.test.ts | 32 +- packages/mcp/tests/shutdown.test.ts | 155 +++ .../mcp/tests/subscription-protocol.test.ts | 233 +++++ packages/mcp/tests/subscriptions.test.ts | 462 +++++++++ packages/mcp/tests/tasks.test.ts | 695 +++++++++++++ packages/mcp/tests/tracing.test.ts | 274 +++++ .../onebot/src/network/http-server-adapter.ts | 19 +- packages/onebot/src/network/http-trace.ts | 21 + .../onebot/tests/http-server-tracing.test.ts | 225 +++++ pnpm-lock.yaml | 98 +- 31 files changed, 7317 insertions(+), 56 deletions(-) create mode 100644 packages/mcp/apps/message-preview.ts create mode 100644 packages/mcp/apps/preview-view.ts create mode 100644 packages/mcp/apps/tsconfig.json create mode 100644 packages/mcp/scripts/build-message-app.mjs create mode 100644 packages/mcp/src/elicitation.ts create mode 100644 packages/mcp/src/message-app.ts create mode 100644 packages/mcp/src/shutdown.ts create mode 100644 packages/mcp/src/subscriptions.ts create mode 100644 packages/mcp/src/tasks.ts create mode 100644 packages/mcp/src/tracing.ts create mode 100644 packages/mcp/tests/elicitation.test.ts create mode 100644 packages/mcp/tests/message-app.test.ts create mode 100644 packages/mcp/tests/protocol-versions.test.ts create mode 100644 packages/mcp/tests/shutdown.test.ts create mode 100644 packages/mcp/tests/subscription-protocol.test.ts create mode 100644 packages/mcp/tests/subscriptions.test.ts create mode 100644 packages/mcp/tests/tasks.test.ts create mode 100644 packages/mcp/tests/tracing.test.ts create mode 100644 packages/onebot/src/network/http-trace.ts create mode 100644 packages/onebot/tests/http-server-tracing.test.ts diff --git a/AGENTS.md b/AGENTS.md index b1466790..b18f0e32 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -77,7 +77,8 @@ Use focused tests while iterating. Before handing off a code change, run the ful - **SDK packaging:** keep the SDK's `NodeNext` build and explicit `.js` relative imports, including type-only exports. Its package tests compile fresh output and check native Node.js imports, consumer declarations, and complete action-map coverage. The SDK action catalog must match OneBot's executable names; `packages/onebot/tests/sdk-actions.test.ts` checks this without adding a runtime dependency on server code. SDK parameter/result contracts remain independent of server types; keep consumer examples under `packages/sdk/tests/fixtures` aligned with the server parser checks in `packages/onebot/tests/sdk-action-params.test.ts`. - **SDK streams:** preserve the single-response client methods and use the additive interfaces in [STREAMING.md](packages/sdk/STREAMING.md) for file transfers. Test framing, cancellation, bounded buffering and incomplete transfers in the SDK; `packages/onebot/tests/sdk-streaming.test.ts` checks real adapter/handler round trips in temporary storage without QQ. - **SDK messaging:** preserve existing string-path and ordinary reply semantics when extending the helpers in [MESSAGING.md](packages/sdk/MESSAGING.md). Filesystem access belongs only in the optional `@snowluma/sdk/node` entry point. Media sends must await upload completion and retain failure context without automatic retries; event waits follow public middleware dispatch and release listeners on every outcome. Keep legacy handler/context types assignable when adding richer SDK-created contexts. -- **MCP agents:** preserve existing text results, business-error envelopes, default search matching and mode gates when extending the tools. Additive structured results, discovery controls, runtime checks, cancellation and progress are documented in [AGENT_FEATURES.md](packages/mcp/AGENT_FEATURES.md). Use isolated HTTP and MCP transports in tests; local runtime inspection must not query a live account. +- **MCP agents:** preserve existing text results, business-error envelopes, default search matching and mode gates when extending the tools. Use the server SDK's `serveStdio` entry to support modern discovery and legacy initialization from the same command; test both with freshly compiled output. Cache hints belong only on bundled catalog discovery, never execution or account data. Keep trace metadata out of action parameters and sensitive data out of diagnostics. Protocol compatibility, tracing and the additive agent features are documented in [AGENT_FEATURES.md](packages/mcp/AGENT_FEATURES.md). Use isolated HTTP and MCP transports in tests; local runtime inspection must not query a live account. +- **MCP execution services:** keep background tasks and event connections opt-in. Persist task state before effects, never replay interrupted writes, and finish durable cleanup before draining the protocol on shutdown. Input continuations must retain their original arguments and execute at most once. Event buffers require explicit account binding, bounded storage and visible gaps. Build the bundled message preview through `packages/mcp/scripts/build-message-app.mjs`; loading a preview must not send a message. Test restart, cancellation and legacy compatibility using isolated transports, including the compiled CLI. - **Protocol codecs:** keep the existing Proton integration in tests so they exercise the same transformations as production. Import real implementations; do not duplicate the algorithm in the test. - **Native coverage:** `pnpm test:native` is separate from default protocol tests and requires compatible native artifacts. A missing artifact or unavailable target is a reported limitation, not permission to bypass the check. - **Live QQ behavior:** fixture tests and a successful API response do not prove delivery or client rendering. State what was tested locally and what still needs a real client. diff --git a/packages/mcp/AGENT_FEATURES.md b/packages/mcp/AGENT_FEATURES.md index 33f73d4d..ff311a98 100644 --- a/packages/mcp/AGENT_FEATURES.md +++ b/packages/mcp/AGENT_FEATURES.md @@ -5,6 +5,249 @@ and read/write gates. No endpoint is contacted by catalog tools or by a local runtime overview. The action catalog remains a bundled snapshot, not a claim about the connected server's capabilities. +## Protocol compatibility and catalog caching + +The same `snowluma-mcp` stdio command accepts MCP `2026-07-28` clients and +older clients using the `initialize` handshake. The official server SDK selects +the protocol for the connection. Modern clients can use `server/discover` and +per-request metadata without initialization; legacy clients keep their existing +configuration. This does not add a publicly listening MCP HTTP endpoint. + +Modern replies carry the protocol's completion and server metadata. Cache hints +depend on whether a response contains static catalog data or private live state: + +| Response | Lifetime | Scope | +| ----------------------------------------------------- | --------------------------------------------- | ------------------------------------------------------------------------------ | +| `tools/list` | Five minutes | `private`; mode and client capabilities affect tool metadata and availability. | +| `resources/list` | Five minutes, or zero when events are enabled | `private`; UI capability and active event resources affect the list. | +| `resources/templates/list` | Five minutes | `public`; the bundled template list is static. | +| `resources/read` for the action catalog or message UI | Five minutes | `public`; these are bundled static assets. | +| `resources/read` for event resources | Zero | `private`; these contain account events and connection state. | + +These are client cache hints, not a server-side result cache. Tool calls, +including the runtime overview, never receive these hints. The SDK omits +modern-only result fields for legacy clients. Unsupported protocol revisions +are rejected before action execution; they do not trigger an action retry. + +## Call tracing + +Every tool invocation gets a W3C trace context. Clients can supply `traceparent` +and optional `tracestate` in the tool request's `_meta`: + +```json +{ + "name": "query_action", + "arguments": { "action": "get_status" }, + "_meta": { + "traceparent": "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01" + } +} +``` + +MCP preserves a valid caller trace ID and creates a new span ID for each call. +Without a valid parent it creates an independent trace. The returned +`_meta.traceparent` identifies that call, including tool errors; the existing +text and structured result envelopes remain unchanged. This works with both +protocol generations. + +Ordinary requests, streams, automatic upload chunks, and upload cleanup forward +the call context in HTTP headers, never in OneBot action parameters. Concurrent +calls keep independent headers. To follow a call, match its traceparent in MCP +stderr diagnostics with SnowLuma's `http_trace` entry, then follow that entry's +existing `req#` through action lifecycle logs. SnowLuma correlation requires +TRACE logging and a valid authenticated HTTP request; the remote trace ID does +not replace its local request ID. No telemetry exporter is installed. + +MCP's call diagnostics contain the known tool name, traceparent, outcome, failure +category and duration, not message contents, tokens, file paths or arbitrary metadata. +`tracestate` is validated with a maximum of 512 characters and 32 members; it is +forwarded but not logged. Invalid trace metadata produces a fixed diagnostic and +does not reject an otherwise valid action. `baggage` is neither forwarded nor +logged. Trace identifiers are correlation data, not authentication or permission. + +## Optional execution services + +Existing configurations keep working. Input completion and message previews are +additive tools; event connections and durable tasks require explicit configuration. +For example, set these variables in the MCP client's server environment, replacing +addresses, credentials and paths with the intended SnowLuma instance: + +```json +{ + "SNOWLUMA_MCP_MODE": "write", + "SNOWLUMA_MCP_ENDPOINT": "http://127.0.0.1:3000/", + "SNOWLUMA_MCP_TOKEN": "", + "SNOWLUMA_MCP_EVENTS_URL": "ws://127.0.0.1:3001/event", + "SNOWLUMA_MCP_TASK_DIR": "/absolute/private/mcp-tasks", + "SNOWLUMA_MCP_UPLOAD_ROOT": "/absolute/allowed/uploads" +} +``` + +The HTTP and WebSocket addresses are independently configured server endpoints; +the WebSocket address is never guessed from the HTTP address. Events reuse +`SNOWLUMA_MCP_TOKEN` in an authorization header. Upload and task directories are +on the MCP host. Omit services that are not needed. Docs mode remains offline. + +## Completing missing parameters + +`query_action_with_input` and `invoke_action_with_input` use the same arguments, +action lookup and mode gates as `query_action` and `invoke_action`. The existing +tools continue to behave exactly as before. The additional tools ask for input +only when a field listed in the action catalog's top-level `required` array is +absent; they do not infer conditional requirements or optional values. + +For example, `invoke_action_with_input` with +`{"action":"send_group_msg","params":{"group_id":20001}}` asks for the +missing message. The form contains one `params_json` string, whose contents are +a JSON object such as `{"message":"Hello"}`. The answer is validated against +the missing fields' catalog schemas before normal action execution. Already +provided fields, the action name and execution options cannot be overwritten; +extra answer fields are rejected. A declined, cancelled or invalid answer does +not execute the action. + +Clients must advertise form elicitation support. Modern clients receive +`resultType: "input_required"` and resume using the SDK's request-state and input +response mechanism. Legacy clients with elicitation support use the SDK's +`elicitation/create` flow. Clients without form support receive a tool error and +can provide complete arguments to the original tools instead. + +Continuation state is signed, tied to the original tool and arguments, expires +after ten minutes and is consumed once before execution. Concurrent or repeated +answers cannot execute the same continuation twice. A runtime holds at most 256 +pending prompts; restart invalidates them rather than replaying work. Completed +arguments bypass the input round entirely. This feature is not a general approval +gate and does not add confirmation prompts to existing action calls. + +## Live event resources + +With `SNOWLUMA_MCP_EVENTS_URL` configured, read and write modes expose +`subscribe_events` and `unsubscribe_events`. The URL must use `ws:` or `wss:` +without embedded credentials, query parameters or fragments. Listing tools or +resources does not connect; only `subscribe_events` opens the event connection. + +```json +{ + "self_id": 10001, + "types": ["message", "notice"], + "group_ids": [20001], + "user_ids": [30001] +} +``` + +Use the expected logged-in account for required `self_id`; obtain it through +`get_login_info` if necessary. Every incoming event must match this account. +An account mismatch or malformed event stops the connection and is surfaced. +The optional types are `message`, `notice` and `request`; omitted types mean all +three. `message_sent` is included in the message category. Group and user filters +are combined with AND, so a group filter excludes events without a matching +`group_id`. Each ID filter accepts at most 100 distinct positive integers. + +The result contains a canonical `snowluma://onebot/events/` resource URI. +Read it with `resources/read`; use `?after=` on subsequent reads to +obtain newer buffered entries. Cursors belong to that one resource. The response +includes `status`, `next_cursor`, `oldest_cursor`, `dropped_events`, `gap` and +`notification_failed`. Reads do not consume events. Only events received after +subscription creation are available; this is not message-history storage. + +Each resource retains at most 256 events and 1 MiB of event data. One runtime +supports at most 16 resources, sharing one account connection. Incoming frames +are limited to 256 KiB. Buffer eviction and disconnection remain visible through +the cursor/gap and status fields; the service does not silently reconnect or +claim missing events were recovered. Release existing resources before changing +accounts or creating a new connection after disconnection. + +For updates, use the **canonical URI without `?after`**: + +- Modern clients use `subscriptions/listen` with `resourceSubscriptions: [uri]` + (the TypeScript SDK exposes `client.listen(...)`). Closing that notification + stream does not release the event buffer; call `unsubscribe_events` explicitly. +- Legacy clients first call `resources/subscribe` with the canonical URI, then + receive `notifications/resources/updated`. `resources/unsubscribe` also releases + that event resource. The explicit unsubscribe tool works for both generations. + +An update notification indicates that the resource changed; fetch the resource +to read its data. Releasing the last resource closes the WebSocket. Closing the +MCP connection releases its event resources and buffers. No event operation +executes a OneBot business action or sends a QQ message. + +## Persistent background transfers + +Set an absolute `SNOWLUMA_MCP_TASK_DIR` and use write mode to enable persistent +tasks. Task tools additionally require MCP `2026-07-28` clients advertising +`extensions: {"io.modelcontextprotocol/tasks": {}}`. Legacy clients keep the +existing synchronous execution tools; this server does not expose the obsolete +`tasks/list` or `tasks/result` protocol methods. + +`invoke_action_background` takes the same arguments as `invoke_action`, but only +accepts catalog actions marked as streams. For example, it can run +`upload_file_stream` with `execution.input_file` inside the configured upload root. +It writes the task record before starting work and returns a `resultType: "task"` +handle. Ordinary actions continue through the existing tools. The current +background transfer entry does not automatically combine with `_with_input`; +provide the required transfer parameters before starting it. + +- `tasks/get` with `{ "taskId": "..." }` returns current state and, when finished, + the saved result or error. A `completed` task can contain a failed business + response: inspect the nested result's diagnostics, `status` and `retcode`. +- `tasks/cancel` requests cooperative cancellation. Cancellation does not undo + remote effects, and a successful completion already in flight may win the race. + Poll the task again for its final state. +- `tasks/update` accepts input responses only for outstanding task input requests. + It is the protocol lifecycle endpoint, not an arbitrary task-editing API. The + current background transfer entry does not itself request interactive input. +- `list_tasks` is an MCP tool returning task metadata without result bodies. It + accepts optional `cursor` and `limit` (1–100; default 50) and returns `nextCursor` + when another page is available. + +Task handles advertise `pollIntervalMs: 1000` and `ttlMs: null`. Results survive +restart and are not automatically expired or deleted. Restart marks unfinished +work failed with an explicit **unknown remote outcome** message; it never +replays the operation. This persists transfer status/results, not resumable +upload or download execution. There are at most four concurrent tasks and a +16 MiB limit on each persisted task record. + +Storage uses an exclusive runtime lease, private directories/files (0700/0600 on +POSIX) and validated records. Corruption, permission errors and write failures +remain errors. The directory is bound to a digest of the effective mode, endpoint +and token; changing that configuration rejects reuse of the directory. Choose a +new private directory for the new configuration and retain the old history. +Credentials are not stored in the scope record, but task results can contain +private account data and should be protected accordingly. + +## Message preview panel + +`preview_message` is available in every mode and only constructs a draft: + +```json +{ + "message_type": "group", + "target_id": 20001, + "text": "Hello", + "reply_to": -123456 +} +``` + +Use `private` for a private conversation. `target_id` must be a positive safe +integer; optional `reply_to` is a non-zero signed 32-bit message ID. The preview +supports literal text and optional quoted replies, not media rendering. Text +that resembles message control codes remains literal text. + +Every client receives the same readable text draft, `can_send` flag and +corresponding `invoke_action` arguments. Only clients advertising the MCP Apps +extension `io.modelcontextprotocol/ui` with MIME type +`text/html;profile=mcp-app` receive the tool's UI metadata and listed UI resource. +The panel is a bundled, self-contained HTML asset built with the official Apps +client. It loads no external scripts, fonts or media, opens no network connection +of its own and displays external text without interpreting it as HTML. A missing +packaged asset is a visible packaging error. + +Loading the panel sends nothing. Its send button calls the existing +`invoke_action` through the host only after the user clicks, and the server's +write gate still applies. Read/docs mode or a host without server-tool support +keeps sending disabled. The panel prevents duplicate clicks and does not retry +failed or uncertain sends; create a new draft only after checking the outcome. +Successful acknowledgement is not proof of delivery or rendering in QQ. + ## Action result schemas and examples `get_action` and `snowluma://onebot/actions` include `returnsSchema` and @@ -49,8 +292,11 @@ and decimal strings such as media `file_size` exactly as documented. ## Structured results -Every tool advertises an `outputSchema` and returns `structuredContent` alongside -the existing `content` text block: +Completed tool results use the advertised `outputSchema` and return +`structuredContent` alongside the existing `content` text block. Modern +`input_required` and background task handles are protocol continuations; inspect +their `resultType` first and obtain the completed result through the relevant +continuation or task lifecycle: ```json { diff --git a/packages/mcp/apps/message-preview.ts b/packages/mcp/apps/message-preview.ts new file mode 100644 index 00000000..f5f2c036 --- /dev/null +++ b/packages/mcp/apps/message-preview.ts @@ -0,0 +1,9 @@ +import { App } from '@modelcontextprotocol/ext-apps'; +import { mountMessagePreview } from './preview-view.js'; + +const app = new App( + { name: 'SnowLuma 消息预览', version: '1.0.0' }, + {}, + { strict: true, autoResize: true }, +); +mountMessagePreview(app, document); diff --git a/packages/mcp/apps/preview-view.ts b/packages/mcp/apps/preview-view.ts new file mode 100644 index 00000000..b3b28e07 --- /dev/null +++ b/packages/mcp/apps/preview-view.ts @@ -0,0 +1,196 @@ +import type { App } from '@modelcontextprotocol/ext-apps'; + +export type PreviewApp = Pick< + App, + | 'ontoolresult' + | 'ontoolcancelled' + | 'onerror' + | 'onclose' + | 'connect' + | 'getHostCapabilities' + | 'callServerTool' +>; + +function record(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + +interface Draft { + message_type: 'group' | 'private'; + target_id: number; + text: string; + reply_to?: number; +} + +function parseDraft(value: unknown): Draft | undefined { + if ( + !record(value) || + (value.message_type !== 'group' && value.message_type !== 'private') || + typeof value.target_id !== 'number' || + !Number.isSafeInteger(value.target_id) || + value.target_id <= 0 || + typeof value.text !== 'string' || + value.text.trim().length === 0 || + (value.reply_to !== undefined && + (typeof value.reply_to !== 'number' || + !Number.isInteger(value.reply_to) || + value.reply_to === 0 || + value.reply_to < -2147483648 || + value.reply_to > 2147483647)) + ) { + return undefined; + } + + return { + message_type: value.message_type, + target_id: value.target_id, + text: value.text, + ...(value.reply_to !== undefined ? { reply_to: value.reply_to } : {}), + }; +} + +export function mountMessagePreview(app: PreviewApp, document: Document): void { + const element = (id: string) => { + const value = document.getElementById(id); + + if (!value) { + throw new Error(`Missing preview element: ${id}`); + } + + return value; + }; + + const target = element('target'); + const body = element('message'); + const reply = element('reply'); + const status = element('status'); + const send = element('send') as HTMLButtonElement; + const cancel = element('cancel') as HTMLButtonElement; + let draft: Draft | undefined; + let canSend = false; + let connected = false; + let attempted = false; + let cancelled = false; + + const update = () => { + send.disabled = !connected || !canSend || !draft || attempted || cancelled; + cancel.disabled = !draft || attempted || cancelled; + }; + + const stop = (message: string) => { + connected = false; + status.textContent = message; + update(); + }; + + app.ontoolresult = (result) => { + // A host may repeat delivery on restoration; never re-arm a sent draft. + if (attempted || cancelled) { + return; + } + + const data = record(result.structuredContent) ? result.structuredContent.result : undefined; + draft = !result.isError && record(data) ? parseDraft(data.draft) : undefined; + canSend = record(data) && data.can_send === true; + + if (!draft) { + target.textContent = '未收到有效草稿'; + body.textContent = ''; + reply.textContent = ''; + status.textContent = '草稿无效,请重新创建。'; + update(); + return; + } + + target.textContent = `${draft.message_type === 'group' ? '群聊' : '私聊'} · ${draft.target_id}`; + body.textContent = draft.text; + reply.textContent = draft.reply_to === undefined ? '' : `引用消息:${draft.reply_to}`; + status.textContent = canSend + ? '消息尚未发送,请核对收件人和正文。' + : '当前连接仅允许预览。'; + update(); + }; + + app.ontoolcancelled = () => { + cancelled = true; + status.textContent = '此草稿已取消。'; + update(); + }; + + app.onerror = () => stop('连接出现错误。请重新打开预览;发送状态需以会话为准。'); + app.onclose = () => stop('连接已断开。请重新打开预览;发送状态需以会话为准。'); + cancel.addEventListener('click', () => { + if (attempted) { + return; + } + + cancelled = true; + status.textContent = '此草稿已取消,未发送消息。'; + update(); + }); + send.addEventListener('click', async () => { + if (!connected || !canSend || !draft || attempted || cancelled) { + return; + } + + attempted = true; + update(); + status.textContent = '正在发送…'; + const message: Array<{ type: string; data: Record }> = []; + + if (draft.reply_to !== undefined) { + message.push({ type: 'reply', data: { id: String(draft.reply_to) } }); + } + + message.push({ type: 'text', data: { text: draft.text } }); + + try { + const result = await app.callServerTool({ + name: 'invoke_action', + arguments: { + action: draft.message_type === 'group' ? 'send_group_msg' : 'send_private_msg', + params: { + [draft.message_type === 'group' ? 'group_id' : 'user_id']: draft.target_id, + message, + }, + }, + }); + const structured = result.structuredContent; + const envelope = record(structured) ? structured.result : undefined; + const diagnostics = record(structured) ? structured.diagnostics : undefined; + + if ( + !result.isError && + record(envelope) && + envelope.status === 'ok' && + envelope.retcode === 0 + ) { + const data = envelope.data; + const messageId = record(data) ? data.message_id : undefined; + status.textContent = + typeof messageId === 'number' && Number.isInteger(messageId) + ? `服务端已接受消息。消息编号:${messageId}` + : '服务端已接受消息。'; + } else if (record(diagnostics) && diagnostics.outcome === 'failed') { + status.textContent = `发送失败:${typeof diagnostics.message === 'string' ? diagnostics.message : '服务端拒绝了请求。'} 如需重试,请创建新草稿。`; + } else { + status.textContent = '发送状态尚未确认。请先检查会话,不要重复发送。'; + } + } catch { + status.textContent = '发送状态尚未确认。请先检查会话,不要重复发送。'; + } + }); + update(); + void app.connect().then( + () => { + connected = app.getHostCapabilities()?.serverTools !== undefined; + + if (!connected) { + status.textContent = '当前客户端仅支持预览,不能从面板发送。'; + } + + update(); + }, + () => stop('无法连接到消息服务。请重新打开预览。'), + ); +} diff --git a/packages/mcp/apps/tsconfig.json b/packages/mcp/apps/tsconfig.json new file mode 100644 index 00000000..2b90ed9c --- /dev/null +++ b/packages/mcp/apps/tsconfig.json @@ -0,0 +1,9 @@ +{ + "extends": "../../../tsconfig.base.json", + "compilerOptions": { + "lib": ["ES2022", "DOM"], + "noEmit": true, + "types": [] + }, + "include": ["*.ts"] +} diff --git a/packages/mcp/package.json b/packages/mcp/package.json index b16585f2..2882cde9 100644 --- a/packages/mcp/package.json +++ b/packages/mcp/package.json @@ -27,9 +27,9 @@ "clean": "node -e \"const fs=require('node:fs');fs.rmSync('dist',{recursive:true,force:true})\"", "gen": "vitest run scripts/gen-catalog.test.ts", "prebuild": "pnpm run clean && pnpm run gen", - "build": "tsc -p tsconfig.json", + "build": "tsc -p tsconfig.json && node scripts/build-message-app.mjs", "prepack": "pnpm run build", - "typecheck": "tsc --noEmit", + "typecheck": "tsc --noEmit && tsc -p apps/tsconfig.json", "test": "vitest run", "start": "node dist/server.js" }, @@ -41,15 +41,20 @@ "qq" ], "dependencies": { - "@modelcontextprotocol/sdk": "^1.32.0" + "@modelcontextprotocol/server": "^2.3.1", + "ws": "^8.21.0" }, "devDependencies": { + "@modelcontextprotocol/client": "^2.3.1", + "@modelcontextprotocol/sdk": "^1.32.0", "@snowluma/onebot": "workspace:*", "@snowluma/proton": "workspace:*", "@types/node": "catalog:", "typescript": "catalog:", "vite": "catalog:", - "vitest": "4.1.11" + "vitest": "4.1.11", + "@types/ws": "^8.18.1", + "@modelcontextprotocol/ext-apps": "^2.0.3" }, "engines": { "node": "^24.22.1 || >=26.12.1" diff --git a/packages/mcp/scripts/build-message-app.mjs b/packages/mcp/scripts/build-message-app.mjs new file mode 100644 index 00000000..905edc93 --- /dev/null +++ b/packages/mcp/scripts/build-message-app.mjs @@ -0,0 +1,48 @@ +import { createHash } from 'node:crypto'; +import { mkdir, writeFile } from 'node:fs/promises'; +import { dirname, join, resolve } from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { build } from 'vite'; + +const packageRoot = fileURLToPath(new URL('../', import.meta.url)); +const style = ` +:root{color-scheme:light dark;font-family:system-ui,sans-serif;color:var(--color-text-primary,#20242d);background:var(--color-background-primary,#fff)} +*{box-sizing:border-box}body{margin:0;padding:24px;max-width:760px}h1{font-size:21px;margin:0 0 18px}#target{font-size:14px;font-weight:650;margin:0 0 14px}#message{white-space:pre-wrap;overflow-wrap:anywhere;font:inherit;line-height:1.6;margin:0;padding:18px;border:1px solid var(--color-border-primary,#d7dce3);border-radius:10px}#reply{font-size:13px;margin:12px 0;color:var(--color-text-secondary,#626b78)}#status{font-size:14px;line-height:1.5;margin:18px 0}footer{display:flex;gap:10px;flex-wrap:wrap}button{font:inherit;font-size:14px;border:1px solid var(--color-border-primary,#bbc4d0);border-radius:8px;padding:10px 16px;cursor:pointer;background:var(--color-background-primary,#fff);color:inherit}#send{background:#2252a5;color:white;border-color:#2252a5}button:disabled{opacity:.55;cursor:default}button:focus-visible{outline:3px solid #8eb7ff;outline-offset:3px}@media(prefers-color-scheme:dark){:root{color:var(--color-text-primary,#e8ecf3);background:var(--color-background-primary,#1b1e24)}#reply{color:var(--color-text-secondary,#b3bdcb)}button{background:var(--color-background-primary,#1b1e24)}} +`.trim(); + +export async function buildMessageApp(outDir = join(packageRoot, 'dist')) { + const result = await build({ + configFile: false, + root: packageRoot, + logLevel: 'error', + build: { + write: false, + sourcemap: false, + lib: { + entry: join(packageRoot, 'apps/message-preview.ts'), + name: 'SnowLumaMessagePreview', + formats: ['iife'], + }, + }, + }); + const outputs = Array.isArray(result) ? result.flatMap((item) => item.output) : result.output; + + if (outputs.length !== 1 || outputs[0].type !== 'chunk') { + throw new Error('Message app must build to a single self-contained script'); + } + + const script = outputs[0].code.replace(/<\/script/gi, '<\\/script'); + const hash = (source) => createHash('sha256').update(source).digest('base64'); + const csp = `default-src 'none'; script-src 'sha256-${hash(script)}'; style-src 'sha256-${hash(style)}'; connect-src 'none'; frame-src 'none'; object-src 'none'; base-uri 'none'; form-action 'none'`; + const html = ` +消息草稿预览 +

消息草稿预览

正在载入草稿…

消息尚未发送。

`; + const filename = join(outDir, 'message-app.html'); + await mkdir(dirname(filename), { recursive: true }); + await writeFile(filename, html, 'utf8'); + return filename; +} + +if (process.argv[1] && resolve(process.argv[1]) === fileURLToPath(import.meta.url)) { + await buildMessageApp(); +} diff --git a/packages/mcp/src/client.ts b/packages/mcp/src/client.ts index 8e2f1065..311642b6 100644 --- a/packages/mcp/src/client.ts +++ b/packages/mcp/src/client.ts @@ -11,6 +11,7 @@ import os from 'node:os'; import path from 'node:path'; import { createHash, randomUUID } from 'node:crypto'; import { ActionCallError } from './results.js'; +import type { TraceContext } from './tracing.js'; const DEFAULT_TIMEOUT_MS = 30_000; const DEFAULT_MAX_STREAM_BYTES = 4 * 1024 * 1024 * 1024; @@ -26,7 +27,7 @@ const MAX_FILE_CHUNKS = 200_000; const RESET_TIMEOUT_MS = 5_000; const MAX_ACTIVE_STREAMS = 4; const INTERNAL_STREAM_CALL = Symbol('snowluma.mcp.internalStreamCall'); -const MCP_UPLOAD_OWNED_FIELDS = [ +export const MCP_UPLOAD_OWNED_FIELDS = [ 'stream_id', 'chunk_data', 'chunk_index', @@ -135,6 +136,18 @@ export interface TransferProgress { export interface CallOptions { signal?: AbortSignal; + trace?: TraceContext; +} + +function requestHeaders( + base: Record, + trace?: TraceContext, +): Record { + return { + ...base, + ...(trace ? { traceparent: trace.traceparent } : {}), + ...(trace?.tracestate ? { tracestate: trace.tracestate } : {}), + }; } export interface StreamCallOptions extends CallOptions { @@ -733,7 +746,7 @@ export function makeHttpClient(opts: HttpClientOptions): ActionClient { signal.throwIfAborted(); const res = await fetchImpl(opts.endpoint, { method: 'POST', - headers, + headers: requestHeaders(headers, options?.trace), body: JSON.stringify({ action, params }), signal, }); @@ -842,6 +855,7 @@ export function makeHttpClient(opts: HttpClientOptions): ActionClient { (uploadParams, signal) => client.callStream(action, uploadParams, { ...(signal ? { signal } : {}), + ...(options.trace ? { trace: options.trace } : {}), [INTERNAL_STREAM_CALL]: true, } as InternalStreamCallOptions), options.onProgress, @@ -867,7 +881,7 @@ export function makeHttpClient(opts: HttpClientOptions): ActionClient { throwIfCancelled(options?.signal, action, 'connection'); res = await fetchImpl(opts.endpoint, { method: 'POST', - headers, + headers: requestHeaders(headers, options?.trace), body: JSON.stringify({ action, params }), signal: requestSignal, }); diff --git a/packages/mcp/src/elicitation.ts b/packages/mcp/src/elicitation.ts new file mode 100644 index 00000000..ddc96601 --- /dev/null +++ b/packages/mcp/src/elicitation.ts @@ -0,0 +1,344 @@ +import { createHash, randomBytes, randomUUID } from 'node:crypto'; +import { + createRequestStateCodec, + fromJsonSchema, + inputRequired, + inputResponse, + type InputRequiredResult, + type ServerContext, + type ServerOptions, + type Tool, +} from '@modelcontextprotocol/server'; +import { asError, type StructuredResult } from './results.js'; + +const INPUT_KEY = 'missing_parameters'; +const INPUT_NAMES = { + query_action_with_input: 'query_action', + invoke_action_with_input: 'invoke_action', +} as const; + +export type InputCompletionTool = keyof typeof INPUT_NAMES; + +export interface InputPreparation { + action: string; + params: Record; + inputSchema: Record; +} + +export interface InputCompletionCallbacks { + capabilities: unknown; + /** Runs the same permission and argument checks as the existing execution tools. */ + preflight: (args: Record) => InputPreparation | StructuredResult; + execute: (args: Record) => Promise; +} + +interface PendingInput { + fingerprint: string; + expiresAt: number; +} + +interface InputState { + challenge: string; +} + +export interface InputCompletion { + verifyRequestState: NonNullable['verify']>; + handle: ( + name: InputCompletionTool, + args: Record, + context: ServerContext, + callbacks: InputCompletionCallbacks, + ) => Promise; + close: () => void; +} + +export function isInputCompletionTool(name: string): name is InputCompletionTool { + return Object.hasOwn(INPUT_NAMES, name); +} + +export function inputCompletionBaseTool( + name: InputCompletionTool, +): 'query_action' | 'invoke_action' { + return INPUT_NAMES[name]; +} + +/** Derive visibility, schemas, annotations and result contracts from the existing tools. */ +export function inputCompletionTools(baseTools: Tool[]): Tool[] { + return baseTools.flatMap((tool) => { + if (tool.name !== 'query_action' && tool.name !== 'invoke_action') { + return []; + } + + return [ + { + ...tool, + name: `${tool.name}_with_input`, + description: `${tool.description} 缺少必填参数时向支持表单的客户端请求补充;补充完成前不执行动作,已有参数不会被覆盖。`, + }, + ]; + }); +} + +function isObject(value: unknown): value is Record { + return value !== null && typeof value === 'object' && !Array.isArray(value); +} + +// Bind every round to the exact tool and original arguments, irrespective of JSON +// object key ordering. Values never appear in the signed token or diagnostics. +function fingerprint(name: string, args: Record): string { + function ordered(value: unknown): unknown { + if (Array.isArray(value)) { + return value.map(ordered); + } + + if (isObject(value)) { + return Object.fromEntries( + Object.keys(value) + .sort() + .map((key) => [key, ordered(value[key])]), + ); + } + + return value; + } + + return createHash('sha256') + .update(JSON.stringify([name, ordered(args)])) + .digest('hex'); +} + +function supportsForm(capabilities: unknown): boolean { + const elicitation = isObject(capabilities) ? capabilities.elicitation : undefined; + + if (!isObject(elicitation)) { + return false; + } + + // The pre-form/url capability shape declared form support with an empty object. + return ( + isObject(elicitation.form) || + (elicitation.form === undefined && elicitation.url === undefined) + ); +} + +/** + * One instance belongs to one configured runtime, shared by its server factory. + * Pending prompts are deliberately transient: a restart invalidates unfinished + * input, and never resumes or retries a side effect automatically. + */ +export function createInputCompletion( + options: { ttlSeconds?: number; maxPending?: number } = {}, +): InputCompletion { + const ttlSeconds = options.ttlSeconds ?? 600; + const maxPending = options.maxPending ?? 256; + + if ( + !Number.isInteger(ttlSeconds) || + ttlSeconds < 1 || + !Number.isInteger(maxPending) || + maxPending < 1 + ) { + throw new RangeError('Input completion limits must be positive integers'); + } + + const codec = createRequestStateCodec({ + key: randomBytes(32), + ttlSeconds, + // HTTP callers must supply validated authInfo. Stdio has one trusted peer. + // The codec hides the binding behind an HMAC, including bearer credentials. + bind: (context) => + JSON.stringify([ + context.mcpReq.method, + context.http?.authInfo?.clientId ?? null, + context.http?.authInfo?.token ?? null, + context.sessionId ?? null, + ]), + }); + const pending = new Map(); + let closed = false; + + return { + verifyRequestState: async (token, context) => { + if (closed) { + throw new Error('closed'); + } + + return codec.verify(token, context); + }, + close() { + closed = true; + pending.clear(); + }, + async handle(name, args, context, callbacks) { + if (closed) { + return asError('补充输入服务已关闭。', 'configuration'); + } + + context.mcpReq.signal.throwIfAborted(); + const prepared = callbacks.preflight(args); + + if ('structuredContent' in prepared) { + return prepared; + } + + const state = context.mcpReq.requestState(); + const required = Array.isArray(prepared.inputSchema.required) + ? prepared.inputSchema.required.filter( + (key): key is string => typeof key === 'string', + ) + : []; + const missing = required.filter((key) => !Object.hasOwn(prepared.params, key)); + const original = fingerprint(name, args); + + for (const [id, entry] of pending) { + if (entry.expiresAt <= Date.now()) { + pending.delete(id); + } + } + + if (state !== undefined) { + if (!isObject(state) || typeof state.challenge !== 'string') { + return asError('补充输入状态无效。'); + } + + const entry = pending.get(state.challenge); + + if (!entry || entry.fingerprint !== original) { + return asError('补充输入已过期、已使用或原始调用已变更,请重新发起调用。'); + } + + // Claim synchronously before validation or execution can yield. A + // replay/concurrent retry can never execute the same answer twice. + pending.delete(state.challenge); + const response = inputResponse(context.mcpReq.inputResponses, INPUT_KEY); + + if (response.kind === 'elicit' && response.action !== 'accept') { + return asError( + '用户拒绝或取消补充输入,动作未执行。', + 'cancelled', + '按用户意愿停止;需要时重新发起调用。', + ); + } + + if ( + response.kind !== 'elicit' || + !isObject(response.content) || + typeof response.content.params_json !== 'string' || + Object.keys(response.content).some((key) => key !== 'params_json') + ) { + return asError('补充输入必须包含 params_json 字符串,动作未执行。'); + } + + let additions: unknown; + + try { + additions = JSON.parse(response.content.params_json); + } catch { + return asError('补充参数不是有效 JSON,动作未执行。'); + } + + if ( + !isObject(additions) || + Object.keys(additions).some((key) => !missing.includes(key)) + ) { + return asError( + '补充参数只能包含本次缺失的必填字段,不能覆盖已有参数或添加其他字段。', + ); + } + + const properties = isObject(prepared.inputSchema.properties) + ? prepared.inputSchema.properties + : {}; + const schema = fromJsonSchema>({ + type: 'object', + properties: Object.fromEntries( + missing.map((key) => [key, properties[key] ?? {}]), + ), + required: missing, + additionalProperties: false, + }); + const validated = await schema['~standard'].validate(additions); + + if (validated.issues) { + return asError( + '补充参数不符合动作目录中的字段类型或约束,动作未执行。', + 'invalid_arguments', + '使用 get_action 查看参数要求后重新发起调用。', + ); + } + + context.mcpReq.signal.throwIfAborted(); + + if (closed) { + return asError('补充输入服务已关闭,动作未执行。', 'configuration'); + } + + return callbacks.execute({ + ...args, + params: { ...prepared.params, ...validated.value }, + }); + } + + if (context.mcpReq.inputResponses || context.mcpReq.droppedInputResponseKeys?.length) { + return asError('补充输入缺少有效请求状态,动作未执行。'); + } + + if (!missing.length) { + return callbacks.execute(args); + } + + if (!supportsForm(callbacks.capabilities)) { + return asError( + '当前客户端未声明表单补充输入能力;请直接提供必填参数。', + 'configuration', + '使用 get_action 查看参数后调用原 query_action 或 invoke_action。', + ); + } + + if (pending.size >= maxPending) { + return asError( + '待补充输入的请求数量已达上限,请完成现有请求或等待过期。', + 'execution', + ); + } + + const challenge = randomUUID(); + pending.set(challenge, { + fingerprint: original, + expiresAt: Date.now() + ttlSeconds * 1000, + }); + + try { + const requestState = await codec.mint({ challenge }, context); + context.mcpReq.signal.throwIfAborted(); + + if (closed) { + throw new Error('Input completion closed'); + } + + const properties = isObject(prepared.inputSchema.properties) + ? prepared.inputSchema.properties + : {}; + return inputRequired({ + inputRequests: { + [INPUT_KEY]: inputRequired.elicit({ + mode: 'form', + message: `请补充 ${prepared.action} 缺失的必填参数:${missing.join('、')}。仅填写这些字段的 JSON 对象,已有参数保持不变。字段要求:${JSON.stringify(Object.fromEntries(missing.map((key) => [key, properties[key] ?? {}])))}`, + requestedSchema: { + type: 'object', + properties: { + params_json: { type: 'string', title: '缺失参数(JSON 对象)' }, + }, + required: ['params_json'], + }, + }), + }, + requestState, + }); + } catch (error) { + pending.delete(challenge); + throw error; + } + }, + }; +} diff --git a/packages/mcp/src/message-app.ts b/packages/mcp/src/message-app.ts new file mode 100644 index 00000000..4898a62d --- /dev/null +++ b/packages/mcp/src/message-app.ts @@ -0,0 +1,157 @@ +import { readFile } from 'node:fs/promises'; +import type { ReadResourceResult, Resource, Tool } from '@modelcontextprotocol/server'; +import { asError, asText, RESULT_SCHEMA, type StructuredResult } from './results.js'; + +export const MESSAGE_PREVIEW_TOOL = 'preview_message'; +export const MESSAGE_APP_URI = 'ui://snowluma/message-preview.html'; +export const MESSAGE_APP_MIME_TYPE = 'text/html;profile=mcp-app'; + +function record(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + +export function supportsMessageApp(capabilities: unknown): boolean { + if (!record(capabilities) || !record(capabilities.extensions)) { + return false; + } + + const ui = capabilities.extensions['io.modelcontextprotocol/ui']; + return ( + record(ui) && Array.isArray(ui.mimeTypes) && ui.mimeTypes.includes(MESSAGE_APP_MIME_TYPE) + ); +} + +export function messagePreviewTool(uiSupported: boolean): Tool { + return { + name: MESSAGE_PREVIEW_TOOL, + description: + '预览一条纯文本群聊或私聊草稿,可附带引用消息编号。只生成草稿,不发送消息;支持面板的客户端可由用户点击发送,仍受写模式限制。文字按字面发送,不解释消息控制代码。', + inputSchema: { + type: 'object', + properties: { + message_type: { type: 'string', enum: ['group', 'private'] }, + target_id: { type: 'integer', minimum: 1, maximum: Number.MAX_SAFE_INTEGER }, + text: { type: 'string', minLength: 1 }, + reply_to: { + type: 'integer', + minimum: -2147483648, + maximum: 2147483647, + not: { const: 0 }, + description: '可选引用消息编号,负值合法。', + }, + }, + required: ['message_type', 'target_id', 'text'], + additionalProperties: false, + }, + outputSchema: RESULT_SCHEMA, + annotations: { readOnlyHint: true, openWorldHint: false }, + ...(uiSupported + ? { _meta: { ui: { resourceUri: MESSAGE_APP_URI, visibility: ['model', 'app'] } } } + : {}), + }; +} + +/** Pure draft construction. Sending remains an ordinary invoke_action call. */ +export function previewMessage(args: Record, canSend: boolean): StructuredResult { + if ( + Object.keys(args).some( + (key) => !['message_type', 'target_id', 'text', 'reply_to'].includes(key), + ) + ) { + return asError('草稿包含不支持的字段。'); + } + + const { message_type: messageType, target_id: targetId, text, reply_to: replyTo } = args; + + if (messageType !== 'group' && messageType !== 'private') { + return asError('message_type 必须为 group 或 private。'); + } + + if (typeof targetId !== 'number' || !Number.isSafeInteger(targetId) || targetId <= 0) { + return asError('target_id 必须是正的安全整数。'); + } + + if (typeof text !== 'string' || text.trim().length === 0) { + return asError('text 必须是非空文本。'); + } + + if ( + replyTo !== undefined && + (typeof replyTo !== 'number' || + !Number.isInteger(replyTo) || + replyTo === 0 || + replyTo < -2147483648 || + replyTo > 2147483647) + ) { + return asError('reply_to 必须是非零的有符号 32 位消息编号,负值合法。'); + } + + const message: Array<{ type: string; data: Record }> = []; + + if (replyTo !== undefined) { + message.push({ type: 'reply', data: { id: String(replyTo) } }); + } + + // Explicit text segments preserve literal control-code-looking text. + message.push({ type: 'text', data: { text } }); + return asText({ + draft: { + message_type: messageType, + target_id: targetId, + text, + ...(replyTo !== undefined ? { reply_to: replyTo } : {}), + }, + can_send: canSend, + notice: '此为待发送草稿,尚未发送。', + invocation: { + name: 'invoke_action', + arguments: { + action: messageType === 'group' ? 'send_group_msg' : 'send_private_msg', + params: { [messageType === 'group' ? 'group_id' : 'user_id']: targetId, message }, + }, + }, + }); +} + +export function messageAppResources(uiSupported: boolean): Resource[] { + return uiSupported + ? [ + { + uri: MESSAGE_APP_URI, + name: '消息草稿预览', + description: '查看消息草稿并按需手动发送。', + mimeType: MESSAGE_APP_MIME_TYPE, + }, + ] + : []; +} + +export async function readMessageAppResource(uri: string): Promise { + if (uri !== MESSAGE_APP_URI) { + return undefined; + } + + // The build emits the same self-contained asset beside this compiled module. + // A missing asset is a packaging error and must not become a blank panel. + const text = await readFile(new URL('./message-app.html', import.meta.url), 'utf8'); + return { + contents: [ + { + uri, + mimeType: MESSAGE_APP_MIME_TYPE, + text, + _meta: { + ui: { + csp: { + connectDomains: [], + resourceDomains: [], + frameDomains: [], + baseUriDomains: [], + }, + prefersBorder: true, + }, + }, + }, + ], + }; +} diff --git a/packages/mcp/src/results.ts b/packages/mcp/src/results.ts index 2114ede5..6507d2b2 100644 --- a/packages/mcp/src/results.ts +++ b/packages/mcp/src/results.ts @@ -1,4 +1,4 @@ -import type { CallToolResult, Tool } from '@modelcontextprotocol/sdk/types.js'; +import type { CallToolResult, Tool } from '@modelcontextprotocol/server'; import type { OneBotEnvelope } from './client.js'; export type FailureKind = diff --git a/packages/mcp/src/server-factory.ts b/packages/mcp/src/server-factory.ts index ea9b7e61..813c8067 100644 --- a/packages/mcp/src/server-factory.ts +++ b/packages/mcp/src/server-factory.ts @@ -1,52 +1,336 @@ -import { Server } from '@modelcontextprotocol/sdk/server/index.js'; import { - CallToolRequestSchema, - ListResourcesRequestSchema, - ListToolsRequestSchema, - ReadResourceRequestSchema, -} from '@modelcontextprotocol/sdk/types.js'; + Server, + ProtocolError, + isInputRequiredResult, + isSpecType, + type ServerContext, + type Tool, +} from '@modelcontextprotocol/server'; import { ACTIONS, CATEGORIES } from './generated/catalog.js'; -import { callTool, computeTools, type CallCtx } from './tools.js'; +import { callTool, computeTools, prepareActionCall, type CallCtx } from './tools.js'; import { createProgressReporter } from './progress.js'; import { VERSION } from './version.js'; +import { createCallTrace } from './tracing.js'; +import { + createInputCompletion, + inputCompletionBaseTool, + inputCompletionTools, + isInputCompletionTool, + type InputCompletion, +} from './elicitation.js'; +import { + MESSAGE_PREVIEW_TOOL, + messageAppResources, + messagePreviewTool, + previewMessage, + readMessageAppResource, + supportsMessageApp, +} from './message-app.js'; +import { EventSubscriptions, EVENT_TOOLS, type EventSubscriptionOptions } from './subscriptions.js'; +import { + installTaskHandlers, + supportsTasks, + type PersistentTasks, + type TaskHandle, +} from './tasks.js'; +import { asError, asText, RESULT_SCHEMA, type StructuredResult } from './results.js'; + +export interface McpServerOptions extends Pick { + inputCompletion?: InputCompletion; + tasks?: PersistentTasks; + events?: EventSubscriptionOptions; +} const RESOURCE_URI = 'snowluma://onebot/actions'; -export function createMcpServer(ctx: Pick): Server { +export function createMcpServer(ctx: McpServerOptions): Server { + const input = ctx.inputCompletion ?? createInputCompletion(); + const events = + ctx.mode !== 'docs' && ctx.events ? new EventSubscriptions(ctx.events) : undefined; + const legacySubscriptions = new Set(); const server = new Server( { name: 'snowluma-mcp', version: VERSION }, - { capabilities: { tools: {}, resources: {} } }, + { + capabilities: { + tools: {}, + resources: events ? { subscribe: true, listChanged: true } : {}, + extensions: { 'io.modelcontextprotocol/ui': {} }, + }, + requestState: { verify: input.verifyRequestState }, + // Tool availability depends on the configured mode; only the bundled + // action catalog is public. Execution and account data are never cached. + cacheHints: { + 'tools/list': { ttlMs: 300_000, cacheScope: 'private' }, + 'resources/list': { ttlMs: events ? 0 : 300_000, cacheScope: 'private' }, + 'resources/read': { ttlMs: 0, cacheScope: 'private' }, + 'resources/templates/list': { ttlMs: 300_000, cacheScope: 'public' }, + }, + }, ); - server.setRequestHandler(ListToolsRequestSchema, async () => ({ - tools: computeTools(ctx.mode), + const capabilities = (request: ServerContext) => { + const envelope: Record | undefined = request.mcpReq.envelope; + + if (!envelope) { + return server.getClientCapabilities(); + } + + const value = envelope['io.modelcontextprotocol/clientCapabilities']; + + if (!isSpecType.ClientCapabilities(value)) { + throw new ProtocolError(-32602, 'Invalid client capabilities'); + } + + return value; + }; + + const cacheMetadata = (scope: 'public' | 'private') => + (server.getNegotiatedProtocolVersion() ?? '') >= '2026-07-28' + ? { ttlMs: scope === 'public' ? 300_000 : 0, cacheScope: scope } + : {}; + const taskEnabled = (request: ServerContext) => + !!ctx.tasks && + ctx.mode === 'write' && + (server.getNegotiatedProtocolVersion() ?? '') >= '2026-07-28' && + supportsTasks(request.mcpReq.envelope); + + const availableTools = (request: ServerContext): Tool[] => { + const base = computeTools(ctx.mode); + const extra = [ + ...inputCompletionTools(base), + messagePreviewTool(supportsMessageApp(capabilities(request))), + ...(events ? EVENT_TOOLS : []), + ]; + + if (taskEnabled(request)) { + const invocation = base.find((tool) => tool.name === 'invoke_action')!; + extra.push( + { + ...invocation, + name: 'invoke_action_background', + description: + '在后台执行一个流式传输动作,返回可跨重启查询的任务编号。仅限流式动作和写模式;重启不会自动重放未完成操作。', + }, + { + name: 'list_tasks', + description: '列出本配置下持久化任务的编号和状态,不返回结果正文。', + inputSchema: { + type: 'object', + properties: { + cursor: { type: 'string' }, + limit: { type: 'integer', minimum: 1, maximum: 100 }, + }, + additionalProperties: false, + }, + outputSchema: RESULT_SCHEMA, + annotations: { readOnlyHint: true, openWorldHint: false }, + }, + ); + } + + return [...base, ...extra]; + }; + + if (ctx.tasks && ctx.mode === 'write') { + installTaskHandlers(server, ctx.tasks); + } + + events?.onUpdate(async (uri) => { + if ( + (server.getNegotiatedProtocolVersion() ?? '') >= '2026-07-28' || + legacySubscriptions.has(uri) + ) { + await server.sendResourceUpdated({ uri }); + } + }); + + server.onclose = () => { + events?.close(); + legacySubscriptions.clear(); + + if (!ctx.inputCompletion) { + input.close(); + } + }; + + server.setRequestHandler('tools/list', async (_req, request) => ({ + tools: availableTools(request), })); - server.setRequestHandler(CallToolRequestSchema, async (req, extra) => { - const token = req.params._meta?.progressToken; + server.setRequestHandler('tools/call', async (req, request) => { + const { mcpReq } = request; + const trace = createCallTrace(mcpReq._meta); + const tool = availableTools(request).some((entry) => entry.name === req.params.name) + ? req.params.name + : 'unlisted'; + const started = performance.now(); + + const report = (event: string, outcome?: string, kind?: string, taskId?: string) => { + console.error( + '[snowluma-mcp] %s', + JSON.stringify({ + event, + tool, + traceparent: trace.traceparent, + ...(outcome + ? { outcome, duration_ms: Math.round(performance.now() - started) } + : {}), + ...(kind ? { kind } : {}), + ...(taskId ? { task_id: taskId } : {}), + }), + ); + }; + + report('call_started'); + const token = mcpReq._meta?.progressToken; const progress = token === undefined ? undefined : createProgressReporter( (value) => - extra.sendNotification({ + mcpReq.notify({ method: 'notifications/progress', params: { progressToken: token, ...value }, }), - extra.signal, + mcpReq.signal, ); - const result = await callTool(req.params.name, req.params.arguments ?? {}, { - ...ctx, - signal: extra.signal, - ...(progress ? { onProgress: progress.report } : {}), - }); - if (!result.isError) { - await progress?.finish(); + try { + const args = req.params.arguments ?? {}; + const callContext: CallCtx = { + mode: ctx.mode, + client: ctx.client, + signal: mcpReq.signal, + trace, + ...(progress ? { onProgress: progress.report } : {}), + }; + + const dispatch = async (): Promise< + StructuredResult | TaskHandle | Awaited> + > => { + if (isInputCompletionTool(req.params.name)) { + const base = inputCompletionBaseTool(req.params.name); + return input.handle(req.params.name, args, request, { + capabilities: capabilities(request), + preflight: (values) => { + const prepared = prepareActionCall( + values, + callContext, + base === 'query_action' ? 'read' : 'write', + ); + return 'structuredContent' in prepared + ? prepared + : { + action: prepared.action, + params: prepared.params, + inputSchema: prepared.inputSchema, + }; + }, + execute: (values) => callTool(base, values, callContext), + }); + } + + if (req.params.name === MESSAGE_PREVIEW_TOOL) { + return previewMessage(args, ctx.mode === 'write' && !!ctx.client); + } + + if (events?.handlesTool(req.params.name)) { + const result = await events.callTool(req.params.name, args, mcpReq.signal); + + if (!result.isError) { + if ( + req.params.name === 'unsubscribe_events' && + typeof args.uri === 'string' + ) { + legacySubscriptions.delete(args.uri); + } + + await server.sendResourceListChanged(); + } + + return result; + } + + if ( + req.params.name === 'invoke_action_background' || + req.params.name === 'list_tasks' + ) { + if (!taskEnabled(request) || !ctx.tasks) { + return asError( + '持久任务需配置任务目录、启用写模式,并由客户端声明新版任务能力。', + 'configuration', + ); + } + + if (req.params.name === 'list_tasks') { + if ( + Object.keys(args).some((key) => key !== 'cursor' && key !== 'limit') || + (args.cursor !== undefined && typeof args.cursor !== 'string') || + (args.limit !== undefined && typeof args.limit !== 'number') + ) { + return asError('任务列表参数无效。'); + } + + return asText( + ctx.tasks.list({ + cursor: args.cursor as string | undefined, + limit: args.limit as number | undefined, + }), + ); + } + + const prepared = prepareActionCall(args, callContext, 'write'); + + if ('structuredContent' in prepared) { + return prepared; + } + + if (!prepared.catalog.stream) { + return asError('后台任务仅支持流式传输动作;普通动作请使用原执行工具。'); + } + + mcpReq.signal.throwIfAborted(); + return ctx.tasks.start(async ({ signal }) => + callTool('invoke_action', args, { + mode: ctx.mode, + client: ctx.client, + signal, + trace, + }), + ); + } + + return callTool(req.params.name, args, callContext); + }; + + const result = await dispatch(); + + if (isInputRequiredResult(result)) { + report('call_input_required', 'unknown'); + return { ...result, _meta: { traceparent: trace.traceparent } }; + } + + if (!('structuredContent' in result)) { + report('call_detached', 'unknown', undefined, result.taskId); + return { ...result, _meta: { traceparent: trace.traceparent } }; + } + + if (!result.isError) { + await progress?.finish(); + } + + const { outcome, kind } = result.structuredContent.diagnostics; + report( + mcpReq.signal.aborted || kind === 'cancelled' ? 'call_cancelled' : 'call_finished', + outcome, + kind, + ); + return { ...result, _meta: { ...result._meta, traceparent: trace.traceparent } }; + } catch (error) { + report(mcpReq.signal.aborted ? 'call_cancelled' : 'call_failed', 'unknown'); + throw error; } - - return result; }); - server.setRequestHandler(ListResourcesRequestSchema, async () => ({ + server.setRequestHandler('resources/list', async (_req, request) => ({ resources: [ { uri: RESOURCE_URI, @@ -54,14 +338,28 @@ export function createMcpServer(ctx: Pick): Server { mimeType: 'application/json', description: `SnowLuma v${VERSION} 的 ${ACTIONS.length} 个 OneBot action 完整目录(文档 + JSON Schema)。`, }, + ...messageAppResources(supportsMessageApp(capabilities(request))), + ...(events?.resources() ?? []), ], })); - server.setRequestHandler(ReadResourceRequestSchema, async (req) => { + server.setRequestHandler('resources/templates/list', async () => ({ resourceTemplates: [] })); + server.setRequestHandler('resources/read', async (req) => { + if (events?.ownsUri(req.params.uri)) { + return { ...events.readResource(req.params.uri), ...cacheMetadata('private') }; + } + + const app = await readMessageAppResource(req.params.uri); + + if (app) { + return { ...app, ...cacheMetadata('public') }; + } + if (req.params.uri !== RESOURCE_URI) { throw new Error(`unknown resource: ${req.params.uri}`); } return { + ...cacheMetadata('public'), contents: [ { uri: RESOURCE_URI, @@ -75,5 +373,28 @@ export function createMcpServer(ctx: Pick): Server { ], }; }); + + if (events) { + server.setRequestHandler('resources/subscribe', async (req) => { + if (!events.resources().some((resource) => resource.uri === req.params.uri)) { + throw new ProtocolError(-32602, 'Subscribe using the canonical event resource URI'); + } + + events.readResource(req.params.uri); + legacySubscriptions.add(req.params.uri); + return {}; + }); + server.setRequestHandler('resources/unsubscribe', async (req) => { + if (!events.ownsUri(req.params.uri)) { + throw new ProtocolError(-32602, 'Unknown event resource'); + } + + events.remove(req.params.uri); + legacySubscriptions.delete(req.params.uri); + await server.sendResourceListChanged(); + return {}; + }); + } + return server; } diff --git a/packages/mcp/src/server.ts b/packages/mcp/src/server.ts index f6fd3c2b..04c561eb 100644 --- a/packages/mcp/src/server.ts +++ b/packages/mcp/src/server.ts @@ -20,10 +20,13 @@ // SNOWLUMA_MCP_STREAM_DIR MCP-host directory for completed downloads. // SNOWLUMA_MCP_UPLOAD_ROOT Existing MCP-host directory allowed for uploads. // SNOWLUMA_MCP_MAX_STREAM_BYTES Per-file cap (1..4 GiB; default 4 GiB). +// SNOWLUMA_MCP_TASK_DIR Private persistent task directory (write mode only). +// SNOWLUMA_MCP_EVENTS_URL Explicit OneBot event WebSocket endpoint. // // NOTE: stdout is the MCP protocol channel — all diagnostics go to stderr. -import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js'; +import { serveStdio, StdioServerTransport } from '@modelcontextprotocol/server/stdio'; +import { createHash } from 'node:crypto'; import { ACTIONS, CATEGORIES } from './generated/catalog.js'; import { makeHttpClient, @@ -33,7 +36,11 @@ import { } from './client.js'; import { parseMode, type Mode } from './tools.js'; -import { createMcpServer } from './server-factory.js'; +import { createMcpServer, type McpServerOptions } from './server-factory.js'; +import { createInputCompletion } from './elicitation.js'; +import { PersistentTasks } from './tasks.js'; +import { shutdownRuntime } from './shutdown.js'; +import { normalizeEventSubscriptionOptions } from './subscriptions.js'; import { VERSION } from './version.js'; // ── resolve mode + client from the environment ──────────────────────────── @@ -74,13 +81,95 @@ function resolveRuntime(): { mode: Mode; client?: ActionClient } { return { mode, client }; } -const { mode, client } = resolveRuntime(); +const runtime: McpServerOptions = resolveRuntime(); +const { mode } = runtime; +const taskDirectory = process.env.SNOWLUMA_MCP_TASK_DIR?.trim(); +const eventsUrl = process.env.SNOWLUMA_MCP_EVENTS_URL?.trim(); -const server = createMcpServer({ mode, client }); +if (taskDirectory && mode !== 'write') { + throw new Error('SNOWLUMA_MCP_TASK_DIR requires write mode and a configured endpoint'); +} + +if (eventsUrl && mode !== 'docs') { + runtime.events = normalizeEventSubscriptionOptions({ + url: eventsUrl, + accessToken: process.env.SNOWLUMA_MCP_TOKEN?.trim() || undefined, + timeoutMs: parseTimeoutMs(process.env.SNOWLUMA_MCP_TIMEOUT_MS), + }); +} + +if (taskDirectory) { + const scopeId = createHash('sha256') + .update( + JSON.stringify({ + mode, + endpoint: process.env.SNOWLUMA_MCP_ENDPOINT?.trim(), + token: process.env.SNOWLUMA_MCP_TOKEN?.trim() || null, + }), + ) + .digest('hex'); + runtime.tasks = await PersistentTasks.open(taskDirectory, { + scopeId, + onChange: (task) => + console.error( + '[snowluma-mcp] %s', + JSON.stringify({ event: 'task_status', task_id: task.taskId, status: task.status }), + ), + onError: (_error, context) => { + console.error('[snowluma-mcp] %s', JSON.stringify(context)); + }, + }); +} + +const inputCompletion = createInputCompletion(); +runtime.inputCompletion = inputCompletion; +let cleanup: Promise | undefined; + +function closeFeatures(): Promise { + return (cleanup ??= (async () => { + inputCompletion.close(); + await runtime.tasks?.close(); + })()); +} + +function reportShutdownError(error: unknown): void { + console.error( + '[snowluma-mcp] shutdown failed: %s', + error instanceof Error ? error.name : 'unknown', + ); + process.exitCode = 1; +} const transport = new StdioServerTransport(); -await server.connect(transport); -const where = client ? ` → ${process.env.SNOWLUMA_MCP_ENDPOINT?.trim()}` : ''; +const handle = serveStdio(() => createMcpServer(runtime), { + transport, + onerror: (error) => { + // Protocol errors may embed client input. Keep the error category visible + // without copying arguments or credentials into process diagnostics. + console.error('[snowluma-mcp] protocol error: %s', error.name); + }, +}); +const transportClosed = transport.onclose; + +transport.onclose = () => { + transportClosed?.(); + void closeFeatures().catch(reportShutdownError); +}; + +for (const signal of ['SIGINT', 'SIGTERM'] as const) { + process.once(signal, () => { + void shutdownRuntime({ + closeFeatures, + closeProtocol: () => handle.close(), + forceClose: async () => { + await transport.close(); + process.stdout.destroy(); + }, + onLateProtocolError: reportShutdownError, + }).catch(reportShutdownError); + }); +} + console.error( - `[snowluma-mcp] v${VERSION} ready — ${ACTIONS.length} actions, ${CATEGORIES.length} categories, mode=${mode}${where}`, + `[snowluma-mcp] v${VERSION} ready — ${ACTIONS.length} actions, ${CATEGORIES.length} categories, mode=${mode}`, ); diff --git a/packages/mcp/src/shutdown.ts b/packages/mcp/src/shutdown.ts new file mode 100644 index 00000000..01e3dc0b --- /dev/null +++ b/packages/mcp/src/shutdown.ts @@ -0,0 +1,89 @@ +export interface ShutdownOptions { + closeFeatures(): Promise; + closeProtocol(): Promise; + forceClose(): Promise; + graceMs?: number; + onTimeout?: () => void; + onLateProtocolError?: (error: unknown) => void; +} + +/** Persist task shutdown before waiting on a peer that may have stopped reading stdout. */ +export async function shutdownRuntime(options: ShutdownOptions): Promise { + const failures: unknown[] = []; + + try { + await options.closeFeatures(); + } catch (error) { + failures.push(error); + } + + const protocol = Promise.resolve() + .then(() => options.closeProtocol()) + .then( + () => ({ kind: 'closed' as const }), + (error: unknown) => ({ kind: 'failed' as const, error }), + ); + let timer!: ReturnType; + const timeout = new Promise<{ kind: 'timeout' }>((resolve) => { + timer = setTimeout(() => resolve({ kind: 'timeout' }), options.graceMs ?? 1000); + }); + const outcome = await Promise.race([protocol, timeout]); + clearTimeout(timer); + + if (outcome.kind === 'failed') { + failures.push(outcome.error); + } + + if (outcome.kind === 'timeout') { + // A protocol rejection can arrive after the bounded shutdown returns. + // Observe the settled outcome too, including races with the deadline. + void protocol.then((late) => { + if (late.kind !== 'failed') { + return; + } + + try { + if (options.onLateProtocolError) { + options.onLateProtocolError(late.error); + } else { + console.error( + '[snowluma-mcp] protocol shutdown failed after timeout: %s', + late.error instanceof Error ? late.error.name : 'unknown', + ); + } + } catch (reportError) { + console.error( + '[snowluma-mcp] late shutdown error reporting failed: %s; original: %s', + reportError instanceof Error ? reportError.name : 'unknown', + late.error instanceof Error ? late.error.name : 'unknown', + ); + } + }); + + try { + if (options.onTimeout) { + options.onTimeout(); + } else { + console.error('[snowluma-mcp] protocol shutdown timed out; closing transport'); + } + } catch (error) { + failures.push(error); + } + } + + if (outcome.kind !== 'closed') { + try { + await options.forceClose(); + } catch (error) { + failures.push(error); + } + } + + if (failures.length === 1) { + throw failures[0]; + } + + if (failures.length > 1) { + throw new AggregateError(failures, 'Runtime shutdown failed'); + } +} diff --git a/packages/mcp/src/subscriptions.ts b/packages/mcp/src/subscriptions.ts new file mode 100644 index 00000000..05fbcdef --- /dev/null +++ b/packages/mcp/src/subscriptions.ts @@ -0,0 +1,722 @@ +import { randomUUID } from 'node:crypto'; +import { + ProtocolError, + type ReadResourceResult, + type Resource, + type Tool, +} from '@modelcontextprotocol/server'; +import WebSocket, { type RawData } from 'ws'; +import { asError, asText, RESULT_SCHEMA, type StructuredResult } from './results.js'; + +class EventInputError extends ProtocolError { + constructor(message: string) { + super(-32602, message); + } +} + +const PREFIX = 'snowluma://onebot/events/'; +const MAX_FRAME_BYTES = 256 * 1024; +const MAX_BUFFER_BYTES = 1024 * 1024; +const MAX_BUFFER_EVENTS = 256; +const MAX_SUBSCRIPTIONS = 16; +const MAX_FILTER_IDS = 100; +type EventType = 'message' | 'notice' | 'request'; +type SourceStatus = 'connecting' | 'connected' | 'disconnected'; +type SourceProblem = + | 'connection_failed' + | 'connection_closed' + | 'connection_timeout' + | 'invalid_event' + | 'account_mismatch' + | 'frame_too_large'; + +interface Filter { + selfId: number; + types: Set; + groupIds?: Set; + userIds?: Set; +} + +interface BufferedEvent { + cursor: number; + event: Record; + bytes: number; +} + +interface Subscription { + uri: string; + filter: Filter; + events: BufferedEvent[]; + bytes: number; + cursor: number; + dropped: number; + status: SourceStatus; + problem?: SourceProblem; + notificationFailed: boolean; + cancelPending?: () => void; +} + +interface Source { + socket: WebSocket; + selfId: number; + ready: Promise; + resolve: () => void; + reject: (error: Error) => void; + verified: boolean; + stopped: boolean; + timer?: ReturnType; + heartbeat?: ReturnType; + awaitingPong: boolean; +} + +export interface EventSubscriptionOptions { + url: string; + accessToken?: string; + timeoutMs?: number; +} + +/** Validate startup configuration without creating resources or contacting an account. */ +export function normalizeEventSubscriptionOptions( + options: EventSubscriptionOptions, +): EventSubscriptionOptions & { timeoutMs: number } { + let url: URL; + + try { + url = new URL(options.url); + } catch { + throw new Error('SNOWLUMA_MCP_EVENTS_URL must be an absolute WebSocket URL'); + } + + if ( + !['ws:', 'wss:'].includes(url.protocol) || + url.username || + url.password || + url.hash || + url.search + ) { + throw new Error( + 'SNOWLUMA_MCP_EVENTS_URL must use ws/wss without credentials, query, or fragment; use SNOWLUMA_MCP_TOKEN for authentication', + ); + } + + if ( + options.timeoutMs !== undefined && + (!Number.isSafeInteger(options.timeoutMs) || options.timeoutMs <= 0) + ) { + throw new Error('Event connection timeout must be a positive integer'); + } + + return { ...options, url: url.href, timeoutMs: options.timeoutMs ?? 30_000 }; +} + +const idSchema = { type: 'integer', minimum: 1, maximum: Number.MAX_SAFE_INTEGER }; +const idsSchema = { + type: 'array', + items: idSchema, + minItems: 1, + maxItems: MAX_FILTER_IDS, + uniqueItems: true, +}; + +export const EVENT_TOOLS: Tool[] = [ + { + name: 'subscribe_events', + description: + '订阅指定登录账号的实时 OneBot 事件,返回可读取的资源。消息、通知、请求可按群和用户筛选。只接收创建后的事件;有限缓存溢出和断线会明确标记缺口。通过资源更新通知获知变化,使用 unsubscribe_events 释放资源。', + inputSchema: { + type: 'object', + properties: { + self_id: { + ...idSchema, + description: + '预期登录账号,必须与事件连接的账号相同。可先用 get_login_info 查询。', + }, + types: { + type: 'array', + items: { type: 'string', enum: ['message', 'notice', 'request'] }, + minItems: 1, + maxItems: 3, + uniqueItems: true, + }, + group_ids: idsSchema, + user_ids: idsSchema, + }, + required: ['self_id'], + additionalProperties: false, + }, + outputSchema: RESULT_SCHEMA, + annotations: { readOnlyHint: false, destructiveHint: false, openWorldHint: true }, + }, + { + name: 'unsubscribe_events', + description: + '停止事件订阅并释放其缓存;最后一个事件订阅结束时断开事件连接。不发送任何 OneBot 业务操作。', + inputSchema: { + type: 'object', + properties: { uri: { type: 'string' } }, + required: ['uri'], + additionalProperties: false, + }, + outputSchema: RESULT_SCHEMA, + annotations: { readOnlyHint: false, destructiveHint: false, openWorldHint: false }, + }, +]; + +function record(value: unknown): value is Record { + return value !== null && typeof value === 'object' && !Array.isArray(value); +} + +function id(value: unknown): value is number { + return typeof value === 'number' && Number.isSafeInteger(value) && value > 0; +} + +function readIds(value: unknown, field: string): Set | undefined { + if (value === undefined) { + return undefined; + } + + if ( + !Array.isArray(value) || + value.length === 0 || + value.length > MAX_FILTER_IDS || + !value.every(id) || + new Set(value).size !== value.length + ) { + throw new EventInputError( + `${field} must contain 1..${MAX_FILTER_IDS} distinct positive integer IDs`, + ); + } + + return new Set(value); +} + +function filter(args: Record): Filter { + if ( + Object.keys(args).some( + (key) => !['self_id', 'types', 'group_ids', 'user_ids'].includes(key), + ) + ) { + throw new EventInputError('Unexpected event subscription parameter'); + } + + if (!id(args.self_id)) { + throw new EventInputError('self_id must be a positive safe integer'); + } + + const types = args.types ?? ['message', 'notice', 'request']; + + if ( + !Array.isArray(types) || + types.length === 0 || + types.length > 3 || + types.some((type) => !['message', 'notice', 'request'].includes(type)) || + new Set(types).size !== types.length + ) { + throw new EventInputError('types must contain distinct message, notice, or request values'); + } + + return { + selfId: args.self_id, + types: new Set(types), + groupIds: readIds(args.group_ids, 'group_ids'), + userIds: readIds(args.user_ids, 'user_ids'), + }; +} + +function eventType(value: unknown): EventType | undefined { + if (value === 'message' || value === 'message_sent') { + return 'message'; + } + + if (value === 'notice' || value === 'request') { + return value; + } + + return undefined; +} + +function hasSafeDepth(value: Record): boolean { + const pending: Array<{ value: unknown; depth: number }> = [{ value, depth: 0 }]; + + while (pending.length > 0) { + const entry = pending.pop()!; + + if (entry.value === null || typeof entry.value !== 'object') { + continue; + } + + if (entry.depth >= 64) { + return false; + } + + for (const child of Object.values(entry.value)) { + pending.push({ value: child, depth: entry.depth + 1 }); + } + } + + return true; +} + +/** Private, bounded event resources for one stdio client. Constructing does not connect. */ +export class EventSubscriptions { + private readonly options: EventSubscriptionOptions; + private readonly subscriptions = new Map(); + private readonly listeners = new Set<(uri: string) => Promise>(); + private readonly pendingNotifications = new Set(); + private notificationTimer?: ReturnType; + private notifying = false; + private source?: Source; + private closed = false; + + constructor(options: EventSubscriptionOptions) { + this.options = normalizeEventSubscriptionOptions(options); + } + + handlesTool(name: string): boolean { + return EVENT_TOOLS.some((tool) => tool.name === name); + } + + async callTool( + name: string, + args: Record, + signal?: AbortSignal, + ): Promise { + try { + if (signal?.aborted) { + return asError('事件订阅操作已取消。', 'cancelled'); + } + + if (name === 'subscribe_events') { + const subscription = await this.create(filter(args), signal); + return asText(this.snapshot(subscription, undefined)); + } + + if (name === 'unsubscribe_events') { + if ( + typeof args.uri !== 'string' || + Object.keys(args).some((key) => key !== 'uri') + ) { + throw new EventInputError('unsubscribe_events requires only a resource uri'); + } + + this.remove(args.uri); + return asText({ uri: args.uri, unsubscribed: true }); + } + + return asError('Unknown event tool', 'not_found'); + } catch (error) { + if (signal?.aborted) { + return asError('事件订阅操作已取消。', 'cancelled'); + } + + if (error instanceof EventInputError) { + return asError(error.message, 'invalid_arguments'); + } + + // Socket errors can contain URLs and server-controlled strings. Keep + // those out of both tool diagnostics and stderr. + console.error('[snowluma-mcp] event subscription establishment failed'); + return asError( + '事件订阅未建立;检查事件地址、凭据、预期账号及服务端状态。', + 'transport', + ); + } + } + + resources(): Resource[] { + return Array.from(this.subscriptions.values(), ({ uri }) => ({ + uri, + name: 'OneBot live events', + mimeType: 'application/json', + description: + '有界实时事件队列。资源 URI 加 ?after=游标 可增量读取;gap=true 表示事件不完整。', + })); + } + + ownsUri(uri: string): boolean { + return uri.startsWith(PREFIX); + } + + readResource(uri: string): ReadResourceResult { + const { subscription, after } = this.lookup(uri); + return { + contents: [ + { + uri, + mimeType: 'application/json', + text: JSON.stringify(this.snapshot(subscription, after)), + }, + ], + }; + } + + /** Legacy unsubscribe and the explicit tool both release the source lease. */ + remove(uri: string): void { + const { subscription, after } = this.lookup(uri); + + if (after !== undefined) { + throw new EventInputError('Use the canonical event resource URI to unsubscribe'); + } + + this.subscriptions.delete(subscription.uri); + subscription.cancelPending?.(); + this.pendingNotifications.delete(subscription.uri); + + if (this.subscriptions.size === 0) { + this.stopSource(); + } + } + + onUpdate(listener: (uri: string) => Promise): () => void { + this.listeners.add(listener); + return () => this.listeners.delete(listener); + } + + close(): void { + this.closed = true; + this.stopSource(); + this.subscriptions.clear(); + this.listeners.clear(); + this.pendingNotifications.clear(); + clearTimeout(this.notificationTimer); + } + + private lookup(uri: string): { subscription: Subscription; after?: number } { + const match = + /^snowluma:\/\/onebot\/events\/([a-f0-9-]{36})(?:\?after=(0|[1-9]\d*))?$/.exec(uri); + + if (!match) { + throw new EventInputError('Invalid event resource URI'); + } + + const subscription = this.subscriptions.get(`${PREFIX}${match[1]}`); + const after = match[2] === undefined ? undefined : Number(match[2]); + + if (!subscription) { + throw new EventInputError('Event resource not found or already unsubscribed'); + } + + if (after !== undefined && (!Number.isSafeInteger(after) || after > subscription.cursor)) { + throw new EventInputError('Event cursor is beyond the available stream'); + } + + return { subscription, after }; + } + + private snapshot(subscription: Subscription, after: number | undefined) { + const oldest = subscription.events[0]?.cursor ?? subscription.cursor + 1; + return { + uri: subscription.uri, + self_id: subscription.filter.selfId, + status: subscription.status, + ...(subscription.problem ? { problem: subscription.problem } : {}), + next_cursor: subscription.cursor, + oldest_cursor: oldest, + dropped_events: subscription.dropped, + gap: subscription.status === 'disconnected' || (after ?? 0) < oldest - 1, + notification_failed: subscription.notificationFailed, + events: subscription.events + .filter((entry) => after === undefined || entry.cursor > after) + .map(({ cursor, event }) => ({ cursor, event })), + }; + } + + private async create(selected: Filter, signal?: AbortSignal): Promise { + if (this.closed) { + throw new EventInputError('Event resources are closed'); + } + + if (this.subscriptions.size >= MAX_SUBSCRIPTIONS) { + throw new EventInputError(`At most ${MAX_SUBSCRIPTIONS} event resources may be active`); + } + + if (this.source && (this.source.selfId !== selected.selfId || this.source.stopped)) { + throw new EventInputError( + 'Release existing event resources before changing accounts or reconnecting', + ); + } + + const subscription: Subscription = { + uri: `${PREFIX}${randomUUID()}`, + filter: selected, + events: [], + bytes: 0, + cursor: 0, + dropped: 0, + status: 'connecting', + notificationFailed: false, + }; + this.subscriptions.set(subscription.uri, subscription); + let onAbort: (() => void) | undefined; + + try { + const source = this.source ?? this.connect(selected.selfId); + await Promise.race([ + source.ready, + new Promise((_resolve, reject) => { + onAbort = () => reject(new Error('Event subscription cancelled')); + subscription.cancelPending = onAbort; + signal?.addEventListener('abort', onAbort, { once: true }); + + if (signal?.aborted) { + onAbort(); + } + }), + ]); + + if ( + source.stopped || + signal?.aborted || + this.subscriptions.get(subscription.uri) !== subscription + ) { + throw new Error('Event connection unavailable'); + } + + subscription.status = 'connected'; + return subscription; + } catch (error) { + this.subscriptions.delete(subscription.uri); + this.pendingNotifications.delete(subscription.uri); + + if (this.subscriptions.size === 0) { + this.stopSource(); + } + + throw error; + } finally { + subscription.cancelPending = undefined; + + if (onAbort) { + signal?.removeEventListener('abort', onAbort); + } + } + } + + private connect(selfId: number): Source { + let resolve!: () => void; + let reject!: (error: Error) => void; + const ready = new Promise((accept, fail) => { + resolve = accept; + reject = fail; + }); + const socket = new WebSocket(this.options.url, { + headers: this.options.accessToken + ? { Authorization: `Bearer ${this.options.accessToken}` } + : undefined, + maxPayload: MAX_FRAME_BYTES, + handshakeTimeout: this.options.timeoutMs, + followRedirects: false, + perMessageDeflate: false, + }); + const source: Source = { + socket, + selfId, + ready, + resolve, + reject, + verified: false, + stopped: false, + awaitingPong: false, + }; + this.source = source; + source.timer = setTimeout( + () => this.failSource(source, 'connection_timeout'), + this.options.timeoutMs, + ); + source.timer.unref(); + socket.on('message', (raw: RawData, binary: boolean) => this.receive(source, raw, binary)); + socket.on('error', (error: Error & { code?: string }) => + this.failSource( + source, + error.code === 'WS_ERR_UNSUPPORTED_MESSAGE_LENGTH' + ? 'frame_too_large' + : 'connection_failed', + ), + ); + socket.on('close', () => this.failSource(source, 'connection_closed')); + socket.on('pong', () => { + source.awaitingPong = false; + }); + return source; + } + + private receive(source: Source, raw: RawData, binary: boolean): void { + if (source.stopped || source !== this.source) { + return; + } + + let event: unknown; + let byteLength: number; + + try { + if (binary) { + throw new Error('Binary event frame'); + } + + const bytes = Array.isArray(raw) + ? Buffer.concat(raw) + : raw instanceof ArrayBuffer + ? Buffer.from(raw) + : raw; + byteLength = bytes.length; + event = JSON.parse(bytes.toString('utf8')); + } catch { + this.failSource(source, 'invalid_event'); + return; + } + + if ( + !record(event) || + !id(event.self_id) || + typeof event.post_type !== 'string' || + !hasSafeDepth(event) + ) { + this.failSource(source, 'invalid_event'); + return; + } + + if (event.self_id !== source.selfId) { + this.failSource(source, 'account_mismatch'); + return; + } + + if (!source.verified) { + source.verified = true; + clearTimeout(source.timer); + source.heartbeat = setInterval(() => { + if (source.stopped) { + return; + } + + if (source.awaitingPong) { + this.failSource(source, 'connection_timeout'); + return; + } + + source.awaitingPong = true; + source.socket.ping(); + }, 30_000); + source.heartbeat.unref(); + source.resolve(); + + for (const subscription of this.subscriptions.values()) { + subscription.status = 'connected'; + this.updated(subscription.uri); + } + } + + const type = eventType(event.post_type); + + if (!type) { + return; + } + + for (const subscription of this.subscriptions.values()) { + const selected = subscription.filter; + + if ( + !selected.types.has(type) || + (selected.groupIds && + (!id(event.group_id) || !selected.groupIds.has(event.group_id))) || + (selected.userIds && (!id(event.user_id) || !selected.userIds.has(event.user_id))) + ) { + continue; + } + + subscription.events.push({ cursor: ++subscription.cursor, event, bytes: byteLength }); + subscription.bytes += byteLength; + + while ( + subscription.events.length > MAX_BUFFER_EVENTS || + subscription.bytes > MAX_BUFFER_BYTES + ) { + const dropped = subscription.events.shift()!; + subscription.bytes -= dropped.bytes; + subscription.dropped++; + } + + this.updated(subscription.uri); + } + } + + private failSource(source: Source, problem: SourceProblem): void { + if (source.stopped) { + return; + } + + source.stopped = true; + clearTimeout(source.timer); + clearInterval(source.heartbeat); + source.reject(new Error('Event source unavailable')); + source.socket.terminate(); + console.error('[snowluma-mcp] event source stopped: %s', problem); + + for (const subscription of this.subscriptions.values()) { + subscription.status = 'disconnected'; + subscription.problem = problem; + this.updated(subscription.uri); + } + } + + private stopSource(): void { + if (!this.source) { + return; + } + + const source = this.source; + source.stopped = true; + clearTimeout(source.timer); + clearInterval(source.heartbeat); + source.reject(new Error('Event source closed')); + source.socket.terminate(); + this.source = undefined; + } + + private updated(uri: string): void { + this.pendingNotifications.add(uri); + + if (this.notificationTimer || this.notifying || this.closed) { + return; + } + + this.notificationTimer = setTimeout(() => { + this.notificationTimer = undefined; + void this.flushNotifications(); + }, 25); + this.notificationTimer.unref(); + } + + private async flushNotifications(): Promise { + this.notifying = true; + const uris = Array.from(this.pendingNotifications); + this.pendingNotifications.clear(); + + try { + for (const uri of uris) { + if (!this.subscriptions.has(uri)) { + continue; + } + + for (const listener of this.listeners) { + try { + await listener(uri); + } catch { + const subscription = this.subscriptions.get(uri); + + if (subscription) { + subscription.notificationFailed = true; + } + + console.error('[snowluma-mcp] event resource notification failed'); + } + } + } + } finally { + this.notifying = false; + + if (!this.closed && this.pendingNotifications.size > 0) { + this.updated(this.pendingNotifications.values().next().value!); + } + } + } +} diff --git a/packages/mcp/src/tasks.ts b/packages/mcp/src/tasks.ts new file mode 100644 index 00000000..49671d78 --- /dev/null +++ b/packages/mcp/src/tasks.ts @@ -0,0 +1,956 @@ +import { randomUUID } from 'node:crypto'; +import { constants } from 'node:fs'; +import { lstat, mkdir, open, readdir, realpath, rename, unlink } from 'node:fs/promises'; +import { hostname } from 'node:os'; +import path from 'node:path'; +import { + ProtocolError, + ProtocolErrorCode, + specTypeSchemas, + type CallToolResult, + type InputRequests, + type InputResponses, + type Server, + type ServerContext, + type StandardSchemaV1, +} from '@modelcontextprotocol/server'; + +// The 2026 Tasks extension is server-directed and has get/update/cancel, +// not the deprecated 2025 tasks/result or tasks/list methods. +export const TASK_EXTENSION = 'io.modelcontextprotocol/tasks'; +export const TASK_CAPABILITIES = { extensions: { [TASK_EXTENSION]: {} } }; +const ID = /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/; +const MAX_RECORD_BYTES = 16 * 1024 * 1024; +const INTERRUPTED = 'Task interrupted; remote outcome is unknown. Work was not replayed.'; +const CANCELLED = 'Task cancelled; cancellation does not undo remote effects.'; +const CANCELLATION_REQUESTED = 'Cancellation requested; remote effects may already have occurred.'; + +interface TaskBase { + taskId: string; + createdAt: string; + lastUpdatedAt: string; + ttlMs: null; + pollIntervalMs: number; + statusMessage?: string; +} + +export type PersistentTask = TaskBase & + ( + | { status: 'working' } + | { status: 'input_required'; inputRequests: InputRequests } + | { status: 'completed'; result: CallToolResult & { resultType: 'complete' } } + | { status: 'failed'; error: { code: number; message: string; data?: unknown } } + | { status: 'cancelled' } + ); + +export type TaskSummary = TaskBase & { status: PersistentTask['status'] }; +export type TaskHandle = TaskSummary & { resultType: 'task'; content: [] }; + +export interface TaskExecutionContext { + taskId: string; + signal: AbortSignal; + requestInput(requests: InputRequests): Promise; +} + +export interface TaskStoreOptions { + /** Stable, non-secret digest of the configured account/endpoint and access mode. */ + scopeId: string; + onChange?: (task: PersistentTask) => void | Promise; + onError?: (error: Error, context: { event: string; taskId?: string }) => void; + maxConcurrent?: number; +} + +interface PendingInput { + outstanding: InputRequests; + responses: InputResponses; + resolve: (responses: InputResponses) => void; + reject: (error: Error) => void; +} + +interface ActiveTask { + controller: AbortController; + usedInputKeys: Set; + input?: PendingInput; +} + +function object(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + +export function supportsTasks(meta?: Readonly>): boolean { + const capabilities = meta?.['io.modelcontextprotocol/clientCapabilities']; + + if (!object(capabilities) || !object(capabilities.extensions)) { + return false; + } + + const extension = capabilities.extensions[TASK_EXTENSION]; + return object(extension) && Object.keys(extension).length === 0; +} + +function requireTasks(ctx: ServerContext, version: string | undefined): void { + if (!version || version < '2026-07-28' || !supportsTasks(ctx.mcpReq.envelope)) { + throw new ProtocolError(-32021, 'The request must declare the Tasks extension capability', { + requiredCapabilities: TASK_CAPABILITIES, + }); + } +} + +function taskId(value: unknown): string { + if (typeof value !== 'string' || !ID.test(value)) { + throw new ProtocolError(ProtocolErrorCode.InvalidParams, 'Invalid task identifier'); + } + + return value; +} + +function summary(task: PersistentTask): TaskSummary { + const { taskId, status, createdAt, lastUpdatedAt, ttlMs, pollIntervalMs, statusMessage } = task; + return { + taskId, + status, + createdAt, + lastUpdatedAt, + ttlMs, + pollIntervalMs, + ...(statusMessage ? { statusMessage } : {}), + }; +} + +function terminal(task: PersistentTask): boolean { + return task.status === 'completed' || task.status === 'failed' || task.status === 'cancelled'; +} + +function errorCode(error: unknown): string | undefined { + return object(error) && typeof error.code === 'string' ? error.code : undefined; +} + +async function readPrivateJson(file: string): Promise { + const stat = await lstat(file); + + if ( + !stat.isFile() || + stat.isSymbolicLink() || + (process.platform !== 'win32' && (stat.mode & 0o077) !== 0) + ) { + throw new Error('Task storage file must be a private regular file'); + } + + if (stat.size > MAX_RECORD_BYTES) { + throw new Error('Task storage record exceeds its size limit'); + } + + const handle = await open(file, constants.O_RDONLY | (constants.O_NOFOLLOW ?? 0)); + + try { + const contents = await handle.readFile('utf8'); + + try { + return JSON.parse(contents); + } catch { + throw new Error('Invalid JSON in task storage'); + } + } finally { + await handle.close(); + } +} + +function validateRecord(value: unknown): PersistentTask { + if (!object(value) || value.version !== 1 || !object(value.task)) { + throw new Error('Invalid task storage record'); + } + + const task = value.task; + taskId(task.taskId); + + if ( + typeof task.createdAt !== 'string' || + !Number.isFinite(Date.parse(task.createdAt)) || + typeof task.lastUpdatedAt !== 'string' || + !Number.isFinite(Date.parse(task.lastUpdatedAt)) || + task.ttlMs !== null || + task.pollIntervalMs !== 1000 || + (task.statusMessage !== undefined && typeof task.statusMessage !== 'string') + ) { + throw new Error('Invalid task storage metadata'); + } + + const base: TaskBase = { + taskId: taskId(task.taskId), + createdAt: task.createdAt, + lastUpdatedAt: task.lastUpdatedAt, + ttlMs: null, + pollIntervalMs: 1000, + ...(typeof task.statusMessage === 'string' ? { statusMessage: task.statusMessage } : {}), + }; + + switch (task.status) { + case 'working': + return { ...base, status: 'working' }; + + case 'input_required': { + // In-flight work is never resumed, but corruption must not be hidden. + const requests = validateInputRequests(task.inputRequests); + return { ...base, status: 'input_required', inputRequests: requests }; + } + + case 'cancelled': + return { ...base, status: 'cancelled' }; + + case 'failed': { + if ( + !object(task.error) || + !Number.isSafeInteger(task.error.code) || + typeof task.error.message !== 'string' + ) { + throw new Error('Invalid persisted task error'); + } + + return { + ...base, + status: 'failed', + error: { + code: Number(task.error.code), + message: task.error.message, + ...(task.error.data !== undefined ? { data: task.error.data } : {}), + }, + }; + } + + case 'completed': { + const validated = specTypeSchemas.CallToolResult['~standard'].validate(task.result); + + if (validated.issues) { + throw new Error('Invalid persisted task result'); + } + + return { + ...base, + status: 'completed', + result: { ...validated.value, resultType: 'complete' }, + }; + } + + default: + throw new Error('Unknown persisted task status'); + } +} + +function validateInputRequests(value: unknown): InputRequests { + if (!object(value) || Object.keys(value).length === 0) { + throw new Error('Task input requests must be a non-empty object'); + } + + const requests: InputRequests = Object.create(null); + + for (const [key, request] of Object.entries(value)) { + if (!object(request)) { + throw new Error('Invalid task input request'); + } + + const schema = + request.method === 'elicitation/create' + ? specTypeSchemas.ElicitRequest + : request.method === 'sampling/createMessage' + ? specTypeSchemas.CreateMessageRequest + : request.method === 'roots/list' + ? specTypeSchemas.ListRootsRequest + : undefined; + const parsed = schema?.['~standard'].validate(request); + + if (!parsed || parsed.issues) { + throw new Error('Invalid task input request'); + } + + requests[key] = parsed.value; + } + + return requests; +} + +/** One runtime owns one private directory; callbacks and credentials are never persisted. */ +export class PersistentTasks { + private readonly records = new Map(); + private readonly active = new Map(); + private queue: Promise = Promise.resolve(); + private notifications: Promise = Promise.resolve(); + private closed = false; + private closingRequested = false; + private leaseReleased = false; + private persistenceError?: Error; + private constructor( + private readonly directory: string, + private readonly lease: string, + private readonly options: TaskStoreOptions, + ) {} + + static async open(directory: string, options: TaskStoreOptions): Promise { + if (!path.isAbsolute(directory) || !/^[a-f0-9]{64}$/.test(options.scopeId)) { + throw new Error( + 'Task storage requires an absolute directory and a SHA-256 scope identifier', + ); + } + + if ( + options.maxConcurrent !== undefined && + (!Number.isSafeInteger(options.maxConcurrent) || options.maxConcurrent < 1) + ) { + throw new Error('Task concurrency must be a positive integer'); + } + + await mkdir(directory, { recursive: true, mode: 0o700 }); + const stat = await lstat(directory); + + if ( + !stat.isDirectory() || + stat.isSymbolicLink() || + (process.platform !== 'win32' && (stat.mode & 0o077) !== 0) + ) { + throw new Error( + 'Task storage directory must be private (0700) and cannot be a symlink', + ); + } + + const resolved = await realpath(directory); + const lease = `.lease-${randomUUID()}.json`; + const store = new PersistentTasks(resolved, lease, options); + + try { + await store.save(lease, { pid: process.pid, host: hostname() }); + + // Publishing our lease before inspecting peers makes concurrent starts + // mutually exclusive. A dead owner's lease is recoverable after a crash. + for (const name of await readdir(resolved)) { + if (!/^\.lease-[a-f0-9-]+\.json$/.test(name) || name === lease) { + continue; + } + + let owner: unknown; + + try { + owner = await readPrivateJson(path.join(resolved, name)); + } catch (error) { + if (errorCode(error) === 'ENOENT') { + continue; + } + + throw error; + } + + if ( + !object(owner) || + !Number.isSafeInteger(owner.pid) || + Number(owner.pid) <= 0 || + owner.host !== hostname() + ) { + throw new Error('Task storage has an invalid or foreign-host lock'); + } + + try { + process.kill(Number(owner.pid), 0); + throw new Error('Task storage is locked by another runtime'); + } catch (error) { + if (errorCode(error) !== 'ESRCH') { + throw error; + } + } + + await unlink(path.join(resolved, name)).catch((error: unknown) => { + if (errorCode(error) !== 'ENOENT') { + throw error; + } + }); + store.report('recovered_stale_lock'); + } + + const entries = await readdir(resolved); + + if (entries.includes('scope.json')) { + const scope = await readPrivateJson(path.join(resolved, 'scope.json')); + + if (!object(scope) || scope.version !== 1 || scope.scopeId !== options.scopeId) { + throw new Error( + 'Task storage belongs to a different account or permission scope', + ); + } + } else { + if (entries.some((name) => name.endsWith('.task.json'))) { + throw new Error('Task storage is missing its scope record'); + } + + await store.save('scope.json', { version: 1, scopeId: options.scopeId }); + } + + for (const name of entries) { + if (!name.endsWith('.task.json')) { + continue; + } + + const id = taskId(name.slice(0, -10)); + let task = validateRecord(await readPrivateJson(path.join(resolved, name))); + + if (task.taskId !== id) { + throw new Error('Task storage identifier mismatch'); + } + + if (!terminal(task)) { + task = { + ...summary(task), + status: 'failed', + lastUpdatedAt: new Date().toISOString(), + statusMessage: INTERRUPTED, + error: { code: -32603, message: INTERRUPTED }, + }; + await store.save(name, { version: 1, task }); + store.report('interrupted_task_not_replayed', id); + } + + store.records.set(id, task); + } + + return store; + } catch (error) { + await unlink(path.join(resolved, lease)).catch((cleanupError: unknown) => { + if (errorCode(cleanupError) !== 'ENOENT') { + throw new AggregateError( + [error, cleanupError], + 'Task startup and lock cleanup failed', + ); + } + }); + throw error; + } + } + + private report(event: string, id?: string): void { + const error = new Error(`Task runtime: ${event}`); + + if (this.options.onError) { + try { + this.options.onError(error, { event, ...(id ? { taskId: id } : {}) }); + } catch { + console.error('[snowluma-mcp] task diagnostic callback failed'); + } + } else { + console.error( + '[snowluma-mcp] %s', + JSON.stringify({ event, ...(id ? { taskId: id } : {}) }), + ); + } + } + + private async save(name: string, data: unknown): Promise { + const serialized = JSON.stringify(data); + + if (Buffer.byteLength(serialized) > MAX_RECORD_BYTES) { + throw new Error('Task record exceeds its size limit'); + } + + const temporary = path.join(this.directory, `.tmp-${randomUUID()}`); + const file = await open(temporary, 'wx', 0o600); + + try { + await file.writeFile(serialized); + await file.sync(); + } finally { + await file.close(); + } + + await rename(temporary, path.join(this.directory, name)); + + // Windows does not support opening directories for fsync. + if (process.platform !== 'win32') { + const directory = await open(this.directory, 'r'); + + try { + await directory.sync(); + } finally { + await directory.close(); + } + } + } + + private serial(operation: () => Promise): Promise { + const next = this.queue.then(operation); + this.queue = next.then( + () => undefined, + () => undefined, + ); + return next; + } + + private async put(task: PersistentTask): Promise { + try { + await this.save(`${task.taskId}.task.json`, { version: 1, task }); + } catch (error) { + this.persistenceError = new Error( + 'Task persistence failed; stop using this runtime and inspect its storage', + { cause: error }, + ); + throw this.persistenceError; + } + + this.records.set(task.taskId, task); + + if (this.options.onChange) { + const snapshot = structuredClone(task); + // Subscription backpressure must not block durable state or cancel. + this.notifications = this.notifications + .then(() => this.options.onChange?.(snapshot)) + .catch(() => { + this.report('task_notification_failed', task.taskId); + }); + } + } + + get(id: string): PersistentTask { + if (this.persistenceError) { + throw this.persistenceError; + } + + const task = this.records.get(taskId(id)); + + if (!task) { + throw new ProtocolError(ProtocolErrorCode.InvalidParams, 'Unknown task identifier'); + } + + return structuredClone(task); + } + + list(options: { cursor?: string; limit?: number } = {}): { + tasks: TaskSummary[]; + nextCursor?: string; + } { + if (this.persistenceError) { + throw this.persistenceError; + } + + const limit = options.limit ?? 50; + + if (!Number.isSafeInteger(limit) || limit < 1 || limit > 100) { + throw new ProtocolError( + ProtocolErrorCode.InvalidParams, + 'Task page size must be between 1 and 100', + ); + } + + const entries = [...this.records.values()].sort( + (a, b) => a.createdAt.localeCompare(b.createdAt) || a.taskId.localeCompare(b.taskId), + ); + const previous = + options.cursor === undefined + ? -1 + : entries.findIndex((task) => task.taskId === taskId(options.cursor)); + + if (options.cursor !== undefined && previous < 0) { + throw new ProtocolError(ProtocolErrorCode.InvalidParams, 'Unknown task cursor'); + } + + const page = entries.slice(previous + 1, previous + 1 + limit); + return { + tasks: page.map(summary), + ...(previous + 1 + page.length < entries.length + ? { nextCursor: page.at(-1)!.taskId } + : {}), + }; + } + + async start( + execute: (context: TaskExecutionContext) => Promise, + ): Promise { + if (this.closingRequested) { + throw new Error('Task runtime is closed'); + } + + const task = await this.serial(async () => { + if (this.closed) { + throw new Error('Task runtime is closed'); + } + + if (this.persistenceError) { + throw this.persistenceError; + } + + if (this.active.size >= (this.options.maxConcurrent ?? 4)) { + throw new Error('Task concurrency limit reached'); + } + + const now = new Date().toISOString(); + const task: PersistentTask = { + taskId: randomUUID(), + status: 'working', + createdAt: now, + lastUpdatedAt: now, + ttlMs: null, + pollIntervalMs: 1000, + }; + await this.put(task); + this.active.set(task.taskId, { + controller: new AbortController(), + usedInputKeys: new Set(), + }); + return task; + }); + // Persist the handle before starting a callback that can have side effects. + void this.run(task.taskId, execute); + return { ...summary(task), resultType: 'task', content: [] }; + } + + private async run( + id: string, + execute: (context: TaskExecutionContext) => Promise, + ): Promise { + // close() owns pending records too; keep them registered until it has + // durably marked their outcome instead of dropping them in finally. + if (this.closingRequested) { + return; + } + + const active = this.active.get(id)!; + + try { + if (this.persistenceError) { + throw this.persistenceError; + } + + if (active.controller.signal.aborted) { + throw active.controller.signal.reason; + } + + const result = await execute({ + taskId: id, + signal: active.controller.signal, + requestInput: (requests) => this.requestInput(id, requests), + }); + const validated = specTypeSchemas.CallToolResult['~standard'].validate(result); + + if (validated.issues) { + throw new Error('Invalid task execution result'); + } + + await this.serial(async () => { + if (this.closed || terminal(this.get(id))) { + return; + } + + // A successful late completion can win a cooperative cancellation. + await this.put({ + ...summary(this.get(id)), + status: 'completed', + statusMessage: undefined, + lastUpdatedAt: new Date().toISOString(), + result: { ...validated.value, resultType: 'complete' }, + }); + }); + } catch (error) { + try { + await this.serial(async () => { + if (this.closed || terminal(this.get(id))) { + return; + } + + const base = { + ...summary(this.get(id)), + lastUpdatedAt: new Date().toISOString(), + }; + await this.put( + active.controller.signal.aborted + ? { ...base, status: 'cancelled', statusMessage: CANCELLED } + : { + ...base, + status: 'failed', + statusMessage: + 'Task execution failed; check the correlated call diagnostics.', + error: { + code: error instanceof ProtocolError ? error.code : -32603, + message: + error instanceof ProtocolError + ? error.message + : 'Task execution failed; remote outcome may be unknown.', + ...(error instanceof ProtocolError && error.data !== undefined + ? { data: error.data } + : {}), + }, + }, + ); + }); + } catch { + this.report('task_persistence_failed', id); + } + + if (!active.controller.signal.aborted) { + this.report('task_execution_failed', id); + } + } finally { + active.input?.reject(new Error('Task execution ended')); + this.active.delete(id); + } + } + + private async requestInput(id: string, requests: InputRequests): Promise { + const validated = validateInputRequests(requests); + const pending = await this.serial(async () => { + const active = this.active.get(id); + + if ( + !active || + active.controller.signal.aborted || + this.closed || + terminal(this.get(id)) + ) { + throw new Error(CANCELLED); + } + + if (active.input) { + throw new Error('Task already has pending input'); + } + + if (Object.keys(validated).some((key) => active.usedInputKeys.has(key))) { + throw new Error('Task input keys must be unique for its entire lifetime'); + } + + await this.put({ + ...summary(this.get(id)), + status: 'input_required', + lastUpdatedAt: new Date().toISOString(), + inputRequests: validated, + }); + let resolve!: (responses: InputResponses) => void; + let reject!: (error: Error) => void; + const promise = new Promise((yes, no) => { + resolve = yes; + reject = no; + }); + active.input = { + outstanding: validated, + responses: Object.create(null), + resolve, + reject, + }; + + for (const key of Object.keys(validated)) { + active.usedInputKeys.add(key); + } + + return { promise }; + }); + return pending.promise; + } + + async update(id: string, responses: unknown): Promise { + if (!object(responses)) { + throw new ProtocolError( + ProtocolErrorCode.InvalidParams, + 'Input responses must be an object', + ); + } + + await this.serial(async () => { + if (this.closed) { + throw new Error('Task runtime is closed'); + } + + const task = this.get(id); + const active = this.active.get(id); + const pending = active?.input; + + if (task.status !== 'input_required' || !pending) { + return; + } + + const accepted: InputResponses = Object.create(null); + + for (const [key, response] of Object.entries(responses)) { + if (!Object.hasOwn(pending.outstanding, key)) { + continue; + } + + const request = pending.outstanding[key]; + + if (!request) { + continue; + } + + const schema = + request.method === 'elicitation/create' + ? specTypeSchemas.ElicitResult + : request.method === 'sampling/createMessage' + ? specTypeSchemas.CreateMessageResult + : specTypeSchemas.ListRootsResult; + const parsed = schema['~standard'].validate(response); + + if (parsed.issues) { + throw new ProtocolError( + ProtocolErrorCode.InvalidParams, + 'Invalid task input response', + ); + } + + accepted[key] = parsed.value; + } + + if (Object.keys(accepted).length === 0) { + return; + } + + const outstanding = { ...pending.outstanding }; + + for (const key of Object.keys(accepted)) { + delete outstanding[key]; + } + + const done = Object.keys(outstanding).length === 0; + await this.put( + done + ? { + ...summary(task), + status: 'working', + lastUpdatedAt: new Date().toISOString(), + } + : { + ...summary(task), + status: 'input_required', + lastUpdatedAt: new Date().toISOString(), + inputRequests: outstanding, + }, + ); + Object.assign(pending.responses, accepted); + pending.outstanding = outstanding; + + if (done) { + active.input = undefined; + pending.resolve(pending.responses); + } + }); + } + + async cancel(id: string): Promise { + await this.serial(async () => { + if (this.closed) { + throw new Error('Task runtime is closed'); + } + + const task = this.get(id); + + if (terminal(task)) { + return; + } + + const active = this.active.get(id); + + if (!active) { + throw new Error('Task execution is unavailable'); + } + + await this.put({ + ...task, + lastUpdatedAt: new Date().toISOString(), + statusMessage: CANCELLATION_REQUESTED, + }); + active.controller.abort(new Error(CANCELLED)); + active.input?.reject(new Error(CANCELLED)); + }); + } + + async close(): Promise { + this.closingRequested = true; + await this.serial(async () => { + if (this.leaseReleased) { + return; + } + + this.closed = true; + const activeTasks = [...this.active.entries()]; + + // Abort all executions even when persistence has failed. Failed writes + // still reject close; a later process will terminate the durable state. + for (const [, active] of activeTasks) { + active.controller.abort(new Error(INTERRUPTED)); + active.input?.reject(new Error(INTERRUPTED)); + } + + let failure: unknown; + + for (const [id] of activeTasks) { + const task = this.records.get(id)!; + + try { + if (!terminal(task)) { + await this.put({ + ...summary(task), + status: 'failed', + lastUpdatedAt: new Date().toISOString(), + statusMessage: INTERRUPTED, + error: { code: -32603, message: INTERRUPTED }, + }); + } + } catch (error) { + failure ??= error; + } + } + + await unlink(path.join(this.directory, this.lease)); + this.leaseReleased = true; + this.active.clear(); + + if (failure) { + throw failure; + } + }); + } +} + +const taskParams: StandardSchemaV1 = { + '~standard': { + version: 1 as const, + vendor: 'snowluma', + validate(value: unknown) { + if (!object(value) || typeof value.taskId !== 'string' || !ID.test(value.taskId)) { + return { issues: [{ message: 'Invalid task identifier' }] }; + } + + return { value: { taskId: value.taskId } }; + }, + }, +}; + +export function installTaskHandlers( + server: Server, + tasks: PersistentTasks, + options: { authorize?: (ctx: ServerContext) => void } = {}, +): void { + const authorize = (ctx: ServerContext) => { + requireTasks(ctx, server.getNegotiatedProtocolVersion()); + options.authorize?.(ctx); + }; + + server.registerCapabilities(TASK_CAPABILITIES); + server.setRequestHandler('tasks/get', { params: taskParams }, (params, ctx) => { + authorize(ctx); + return { ...tasks.get(params.taskId) }; + }); + server.setRequestHandler('tasks/cancel', { params: taskParams }, async (params, ctx) => { + authorize(ctx); + await tasks.cancel(params.taskId); + return {}; + }); + server.setRequestHandler('tasks/update', { params: taskParams }, async (params, ctx) => { + authorize(ctx); + const task = tasks.get(params.taskId); + const invalidOutstanding = + task.status === 'input_required' && + ctx.mcpReq.droppedInputResponseKeys?.some((key) => + Object.hasOwn(task.inputRequests, key), + ); + + if (!ctx.mcpReq.inputResponses || invalidOutstanding) { + throw new ProtocolError( + ProtocolErrorCode.InvalidParams, + 'Invalid task input responses', + ); + } + + await tasks.update(params.taskId, ctx.mcpReq.inputResponses); + return {}; + }); +} diff --git a/packages/mcp/src/tools.ts b/packages/mcp/src/tools.ts index 5bf258ce..b4172b84 100644 --- a/packages/mcp/src/tools.ts +++ b/packages/mcp/src/tools.ts @@ -8,10 +8,11 @@ // Visibility is gated by `mode`, and `callTool` RE-checks every permission // (defense in depth: a client may call a tool that was never listed). -import type { Tool } from '@modelcontextprotocol/sdk/types.js'; +import type { Tool } from '@modelcontextprotocol/server'; import { ACTIONS, CATEGORIES } from './generated/catalog.js'; import type { CatalogAction } from './types.js'; import type { ActionClient } from './client.js'; +import { MCP_UPLOAD_OWNED_FIELDS } from './client.js'; import { discover, FILTER_PROPERTIES } from './discovery.js'; import { asText, @@ -24,6 +25,7 @@ import { } from './results.js'; import { VERSION } from './version.js'; import type { TransferProgress } from './client.js'; +import type { TraceContext } from './tracing.js'; export type Mode = 'docs' | 'read' | 'write'; @@ -186,6 +188,7 @@ export interface CallCtx { mode: Mode; client?: ActionClient; signal?: AbortSignal; + trace?: TraceContext; onProgress?: (progress: TransferProgress) => Promise; } @@ -232,13 +235,21 @@ export async function callTool( } } -/** Shared execution path for query_action (read) / invoke_action (write). - * Re-checks mode + readOnly routing regardless of which tools were listed. */ -async function execute( +export interface PreparedActionCall { + action: string; + params: Record; + inputFile?: string; + catalog: CatalogAction; + client: ActionClient; + inputSchema: Record; +} + +/** The same side-effect-free gate is used before execution and input collection. */ +export function prepareActionCall( args: Record, ctx: CallCtx, kind: 'read' | 'write', -): Promise { +): PreparedActionCall | ToolResult { if (kind === 'write' && ctx.mode !== 'write') { return asError( '写操作未启用:需设置 SNOWLUMA_MCP_MODE=write。', @@ -321,22 +332,54 @@ async function execute( return asError('execution.input_file 仅可用于 upload_file_stream。'); } + const inputSchema = + inputFile && Array.isArray(cat.inputSchema.required) + ? { + ...cat.inputSchema, + required: cat.inputSchema.required.filter( + (key) => !MCP_UPLOAD_OWNED_FIELDS.includes(key), + ), + } + : cat.inputSchema; + return { action, params, inputFile, catalog: cat, client: ctx.client, inputSchema }; +} + +/** Shared execution path for query_action (read) / invoke_action (write). */ +async function execute( + args: Record, + ctx: CallCtx, + kind: 'read' | 'write', +): Promise { + const prepared = prepareActionCall(args, ctx, kind); + + if ('structuredContent' in prepared) { + return prepared; + } + + const { action, params, inputFile, catalog: cat, client } = prepared; + try { ctx.signal?.throwIfAborted(); if (cat.stream) { - const result = await ctx.client.callStream(action, params, { + const result = await client.callStream(action, params, { ...(inputFile ? { inputFile } : {}), ...(ctx.signal ? { signal: ctx.signal } : {}), + ...(ctx.trace ? { trace: ctx.trace } : {}), ...(ctx.onProgress ? { onProgress: ctx.onProgress } : {}), }); return asText(result, envelopeDiagnostics(result.terminal)); } - const result = await ctx.client.call( + const result = await client.call( action, params, - ctx.signal ? { signal: ctx.signal } : undefined, + ctx.signal || ctx.trace + ? { + ...(ctx.signal ? { signal: ctx.signal } : {}), + ...(ctx.trace ? { trace: ctx.trace } : {}), + } + : undefined, ); return asText(result, envelopeDiagnostics(result)); } catch (err) { diff --git a/packages/mcp/src/tracing.ts b/packages/mcp/src/tracing.ts new file mode 100644 index 00000000..ed9e7a11 --- /dev/null +++ b/packages/mcp/src/tracing.ts @@ -0,0 +1,110 @@ +import { randomBytes } from 'node:crypto'; + +export interface TraceContext { + traceparent: string; + tracestate?: string; +} + +function parentFields(value: unknown): { traceId: string; flags: string } | undefined { + if (typeof value !== 'string' || /[^\x21-\x7e]/.test(value)) { + return undefined; + } + + const match = /^([0-9a-f]{2})-([0-9a-f]{32})-([0-9a-f]{16})-([0-9a-f]{2})(.*)$/.exec(value); + + if ( + !match || + match[1] === 'ff' || + /^0+$/.test(match[2]!) || + /^0+$/.test(match[3]!) || + (match[1] === '00' ? match[5] !== '' : match[5] !== '' && !match[5]!.startsWith('-')) + ) { + return undefined; + } + + // Participate using version 00, retaining only its defined sampled flag. + return { traceId: match[2]!, flags: (Number.parseInt(match[4]!, 16) & 1) === 1 ? '01' : '00' }; +} + +function traceState(value: unknown): string | undefined { + if (typeof value !== 'string' || value.length > 512 || /[^\x20-\x7e\t]/.test(value)) { + return undefined; + } + + const members = value.split(','); + + if (members.length > 32) { + return undefined; + } + + const keys = new Set(); + const normalized: string[] = []; + + for (const raw of members) { + const member = raw.replace(/^[ \t]+|[ \t]+$/g, ''); + + if (!member) { + continue; + } + + const separator = member.indexOf('='); + const key = member.slice(0, separator); + const entry = member.slice(separator + 1); + + if ( + separator < 1 || + !/^(?:[a-z][a-z0-9_*/-]{0,255}|[a-z0-9][a-z0-9_*/-]{0,240}@[a-z][a-z0-9_*/-]{0,13})$/.test( + key, + ) || + !/^[\x20-\x2b\x2d-\x3c\x3e-\x7e]{0,255}[\x21-\x2b\x2d-\x3c\x3e-\x7e]$/.test(entry) || + keys.has(key) + ) { + return undefined; + } + + keys.add(key); + normalized.push(member); + } + + return normalized.join(','); +} + +function identifier(bytes: number): string { + let value: string; + + do { + value = randomBytes(bytes).toString('hex'); + } while (/^0+$/.test(value)); + + return value; +} + +/** Trace metadata is advisory: invalid fields never reject a tool or enter its params. */ +export function createCallTrace(meta?: Record): TraceContext { + const parent = parentFields(meta?.traceparent); + + if (meta?.traceparent !== undefined && !parent) { + console.error('[snowluma-mcp] trace metadata ignored: invalid traceparent'); + } + + let state: string | undefined; + + if (meta?.tracestate !== undefined) { + if (parent) { + state = traceState(meta.tracestate); + + if (state === undefined) { + console.error('[snowluma-mcp] trace metadata ignored: invalid tracestate'); + } + } else { + console.error( + '[snowluma-mcp] trace metadata ignored: tracestate without valid traceparent', + ); + } + } + + return { + traceparent: `00-${parent?.traceId ?? identifier(16)}-${identifier(8)}-${parent?.flags ?? '00'}`, + ...(state ? { tracestate: state } : {}), + }; +} diff --git a/packages/mcp/tests/elicitation.test.ts b/packages/mcp/tests/elicitation.test.ts new file mode 100644 index 00000000..295f3566 --- /dev/null +++ b/packages/mcp/tests/elicitation.test.ts @@ -0,0 +1,585 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { Client, StreamableHTTPClientTransport } from '@modelcontextprotocol/client'; +import { createMcpHandler, Server, type ClientCapabilities } from '@modelcontextprotocol/server'; +import { Client as LegacyClient } from '@modelcontextprotocol/sdk/client/index.js'; +import { InMemoryTransport } from '@modelcontextprotocol/sdk/inMemory.js'; +import { ElicitRequestSchema } from '@modelcontextprotocol/sdk/types.js'; +import { + createInputCompletion, + inputCompletionBaseTool, + inputCompletionTools, + isInputCompletionTool, +} from '../src/elicitation'; +import { asError, asText } from '../src/results'; +import { computeTools, prepareActionCall, type Mode } from '../src/tools'; +import type { ActionClient } from '../src/client'; +import { createMcpServer } from '../src/server-factory'; + +const cleanup: Array<() => void | Promise> = []; +const completeCapabilities: ClientCapabilities = { elicitation: { form: {} } }; +const ORIGINAL = { action: 'send_private_msg', params: { message: 'private original message' } }; + +afterEach(async () => { + for (const close of cleanup.splice(0).reverse()) { + await close(); + } + + vi.useRealTimers(); + vi.restoreAllMocks(); +}); + +function harness(options: Parameters[0] = {}, mode: Mode = 'write') { + const completion = createInputCompletion(options); + const execute = vi.fn(async (args: Record) => asText(args.params)); + const client: ActionClient = { + call: vi.fn(async () => ({ status: 'ok', retcode: 0 })), + callStream: vi.fn(async () => ({ + frame_count: 1, + frames: [], + terminal: { status: 'ok', retcode: 0 }, + })), + }; + const preflight = vi.fn((args: Record, kind: 'read' | 'write') => { + const result = prepareActionCall(args, { mode, client }, kind); + return 'structuredContent' in result + ? result + : { + action: result.action, + params: result.params, + inputSchema: result.inputSchema, + }; + }); + + const factory = () => { + const server = new Server( + { name: 'input-test', version: '1' }, + { + capabilities: { tools: {} }, + requestState: { verify: completion.verifyRequestState }, + }, + ); + server.setRequestHandler('tools/list', async () => ({ + tools: inputCompletionTools(computeTools('write')), + })); + server.setRequestHandler('tools/call', (request, context) => { + if (!isInputCompletionTool(request.params.name)) { + return asError('unknown tool'); + } + + const name = request.params.name; + return completion.handle(name, request.params.arguments ?? {}, context, { + capabilities: + context.mcpReq.envelope?.['io.modelcontextprotocol/clientCapabilities'] ?? + server.getClientCapabilities(), + preflight: (args) => + preflight( + args, + inputCompletionBaseTool(name) === 'query_action' ? 'read' : 'write', + ), + execute, + }); + }); + return server; + }; + + const handler = createMcpHandler(factory); + cleanup.push( + () => completion.close(), + () => handler.close(), + ); + + async function raw( + arguments_: Record = ORIGINAL, + retry: Record = {}, + capabilities: ClientCapabilities = completeCapabilities, + name = 'invoke_action_with_input', + token?: string, + ) { + const response = await handler.fetch( + new Request('http://input.invalid/mcp', { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + Accept: 'application/json, text/event-stream', + 'Mcp-Method': 'tools/call', + 'Mcp-Name': name, + 'Mcp-Protocol-Version': '2026-07-28', + }, + body: JSON.stringify({ + jsonrpc: '2.0', + id: 1, + method: 'tools/call', + params: { + name, + arguments: arguments_, + ...retry, + _meta: { + 'io.modelcontextprotocol/protocolVersion': '2026-07-28', + 'io.modelcontextprotocol/clientCapabilities': capabilities, + 'io.modelcontextprotocol/clientInfo': { + name: 'input-test', + version: '1', + }, + }, + }, + }), + }), + token ? { authInfo: { token, clientId: 'test-app', scopes: [] } } : undefined, + ); + return response.json(); + } + + return { completion, handler, factory, execute, preflight, raw }; +} + +function answer(requestState: string, content: unknown = { params_json: '{"user_id":123}' }) { + return { requestState, inputResponses: { missing_parameters: { action: 'accept', content } } }; +} + +describe('optional action input completion', () => { + it('integrates with the production factory and preserves the original action result contract', async () => { + const completion = createInputCompletion(); + const onebot: ActionClient = { + call: vi.fn(async () => ({ status: 'ok', retcode: 0, data: { message_id: -42 } })), + callStream: vi.fn(), + }; + const log = vi.spyOn(console, 'error').mockImplementation(() => {}); + const handler = createMcpHandler(() => + createMcpServer({ mode: 'write', client: onebot, inputCompletion: completion }), + ); + const client = new Client( + { name: 'production-input-test', version: '1' }, + { + capabilities: completeCapabilities, + versionNegotiation: { mode: { pin: '2026-07-28' } }, + }, + ); + client.setRequestHandler('elicitation/create', async () => { + expect(onebot.call).not.toHaveBeenCalled(); + return { action: 'accept', content: { params_json: '{"user_id":123}' } }; + }); + cleanup.push( + () => completion.close(), + () => handler.close(), + () => client.close(), + ); + await client.connect( + new StreamableHTTPClientTransport(new URL('http://input.invalid/mcp'), { + fetch: (input, init) => handler.fetch(new Request(input, init)), + }), + ); + const result = await client.callTool({ + name: 'invoke_action_with_input', + arguments: ORIGINAL, + }); + expect(result.structuredContent).toEqual({ + result: { status: 'ok', retcode: 0, data: { message_id: -42 } }, + diagnostics: { outcome: 'success' }, + }); + expect(onebot.call).toHaveBeenCalledExactlyOnceWith( + 'send_private_msg', + { user_id: 123, message: ORIGINAL.params.message }, + expect.objectContaining({ signal: expect.any(AbortSignal) }), + ); + expect(JSON.stringify(log.mock.calls)).not.toContain(ORIGINAL.params.message); + expect(log.mock.calls.map(([, value]) => JSON.parse(String(value)).event)).toEqual([ + 'call_started', + 'call_input_required', + 'call_started', + 'call_finished', + ]); + }); + + it.each(['docs', 'read'] as const)( + 'keeps the production factory mode gate before input collection in %s mode', + async (mode) => { + const onebot: ActionClient = { call: vi.fn(), callStream: vi.fn() }; + const log = vi.spyOn(console, 'error').mockImplementation(() => {}); + const handler = createMcpHandler(() => createMcpServer({ mode, client: onebot })); + const client = new Client( + { name: 'permission-input-test', version: '1' }, + { + capabilities: completeCapabilities, + versionNegotiation: { mode: { pin: '2026-07-28' } }, + }, + ); + const ask = vi.fn(async () => ({ + action: 'accept' as const, + content: { params_json: '{"user_id":123}' }, + })); + client.setRequestHandler('elicitation/create', ask); + cleanup.push( + () => handler.close(), + () => client.close(), + ); + await client.connect( + new StreamableHTTPClientTransport(new URL('http://input.invalid/mcp'), { + fetch: (input, init) => handler.fetch(new Request(input, init)), + }), + ); + const result = await client.callTool({ + name: 'invoke_action_with_input', + arguments: ORIGINAL, + }); + expect(result.structuredContent).toMatchObject({ diagnostics: { kind: 'permission' } }); + expect(ask).not.toHaveBeenCalled(); + expect(onebot.call).not.toHaveBeenCalled(); + expect(log.mock.calls.map(([, value]) => JSON.parse(String(value)).event)).toEqual([ + 'call_started', + 'call_finished', + ]); + }, + ); + + it('inherits tool visibility and existing schemas without altering the original list', () => { + expect(inputCompletionTools(computeTools('docs'))).toEqual([]); + expect(inputCompletionTools(computeTools('read')).map((tool) => tool.name)).toEqual([ + 'query_action_with_input', + ]); + const tools = computeTools('write'); + expect(inputCompletionTools(tools).map((tool) => tool.name)).toEqual([ + 'query_action_with_input', + 'invoke_action_with_input', + ]); + expect(tools.some((tool) => tool.name.endsWith('_with_input'))).toBe(false); + expect(inputCompletionTools(tools)[1].inputSchema).toEqual( + tools.find((tool) => tool.name === 'invoke_action')?.inputSchema, + ); + }); + + it('asks before executing, never discloses existing values, and merges only validated missing fields', async () => { + const { raw, execute } = harness(); + const first = await raw(); + expect(first.result.resultType).toBe('input_required'); + expect(execute).not.toHaveBeenCalled(); + expect(first.result.inputRequests.missing_parameters).toMatchObject({ + method: 'elicitation/create', + }); + expect(JSON.stringify(first)).not.toContain(ORIGINAL.params.message); + const decodedToken = Buffer.from( + first.result.requestState.split('.')[1], + 'base64url', + ).toString(); + expect(decodedToken).not.toContain(ORIGINAL.params.message); + const result = await raw(ORIGINAL, answer(first.result.requestState)); + expect(result.result).toMatchObject({ + resultType: 'complete', + structuredContent: { result: { user_id: 123, message: ORIGINAL.params.message } }, + }); + expect(execute).toHaveBeenCalledExactlyOnceWith({ + action: ORIGINAL.action, + params: { ...ORIGINAL.params, user_id: 123 }, + }); + }); + + it('executes complete arguments without requiring elicitation capability or prompting', async () => { + const { raw, execute } = harness(); + const result = await raw({ action: 'get_msg', params: { message_id: -123 } }, {}, {}); + expect(result.result.structuredContent.result).toEqual({ message_id: -123 }); + expect(execute).toHaveBeenCalledOnce(); + }); + + it('honors permission failure before asking for input or executing', async () => { + const { raw, execute, preflight } = harness(); + preflight.mockReturnValue(asError('write mode required', 'permission')); + const result = await raw(); + expect(result.result.structuredContent.diagnostics.kind).toBe('permission'); + expect(result.result.inputRequests).toBeUndefined(); + expect(execute).not.toHaveBeenCalled(); + }); + + it.each(['docs', 'read'] as const)( + 'enforces actual write permission in %s mode', + async (mode) => { + const { raw, execute } = harness({}, mode); + const result = await raw(); + expect(result.result.structuredContent.diagnostics.kind).toBe('permission'); + expect(result.result.inputRequests).toBeUndefined(); + expect(execute).not.toHaveBeenCalled(); + }, + ); + + it('does not turn a write action into a read action through the input helper', async () => { + const { raw, execute } = harness(); + const result = await raw(ORIGINAL, {}, completeCapabilities, 'query_action_with_input'); + expect(result.result.structuredContent.diagnostics.kind).toBe('permission'); + expect(result.result.inputRequests).toBeUndefined(); + expect(execute).not.toHaveBeenCalled(); + }); + + it('fails visibly when the client has no form capability', async () => { + const { raw, execute } = harness(); + + for (const capabilities of [{}, { elicitation: { url: {} } }]) { + const result = await raw(ORIGINAL, {}, capabilities); + expect(result.result.structuredContent.diagnostics.kind).toBe('configuration'); + } + + expect(execute).not.toHaveBeenCalled(); + }); + + it.each(['decline', 'cancel'])( + 'stops on %s instead of repeatedly prompting', + async (action) => { + const { raw, execute } = harness(); + const first = await raw(); + const result = await raw(ORIGINAL, { + requestState: first.result.requestState, + inputResponses: { missing_parameters: { action } }, + }); + expect(result.result.structuredContent.diagnostics.kind).toBe('cancelled'); + expect(result.result.inputRequests).toBeUndefined(); + expect(execute).not.toHaveBeenCalled(); + }, + ); + + it.each([ + { params_json: '{' }, + { params_json: '{}' }, + { params_json: '{"user_id":0}' }, + { params_json: '{"user_id":"123"}' }, + { params_json: '{"user_id":123,"message":"replacement"}' }, + { params_json: '{"user_id":123,"auto_escape":true}' }, + { params_json: '[]' }, + { params_json: '{"user_id":123}', unexpected: 'value' }, + ])('rejects invalid or extraneous answers without executing: %j', async (content) => { + const { raw, execute } = harness(); + const first = await raw(); + const result = await raw(ORIGINAL, answer(first.result.requestState, content)); + expect(result.result.isError).toBe(true); + expect(execute).not.toHaveBeenCalled(); + }); + + it('accepts complex missing message data without guessing or normalizing it', async () => { + const { raw, execute } = harness(); + const args = { action: 'send_private_msg', params: { user_id: 123 } }; + const message = [ + { type: 'reply', data: { id: -42 } }, + { type: 'text', data: { text: 'hello' } }, + ]; + const first = await raw(args); + await raw( + args, + answer(first.result.requestState, { params_json: JSON.stringify({ message }) }), + ); + expect(execute).toHaveBeenCalledExactlyOnceWith({ + action: args.action, + params: { user_id: 123, message }, + }); + }); + + it('rejects tampered, mismatched and state-less retries', async () => { + const { raw, execute } = harness(); + const first = await raw(); + const tampered = await raw(ORIGINAL, answer(`${first.result.requestState}x`)); + expect(tampered.error.code).toBe(-32602); + const changed = await raw( + { ...ORIGINAL, params: { message: 'different message' } }, + answer(first.result.requestState), + ); + expect(changed.result.isError).toBe(true); + const crossed = await raw( + ORIGINAL, + answer(first.result.requestState), + completeCapabilities, + 'query_action_with_input', + ); + expect(crossed.result.isError).toBe(true); + const noState = await raw(ORIGINAL, { inputResponses: answer('').inputResponses }); + expect(noState.result.isError).toBe(true); + expect(execute).not.toHaveBeenCalled(); + }); + + it('binds retries to the validated caller, including users sharing the same OAuth client', async () => { + const { raw, execute } = harness(); + const first = await raw( + ORIGINAL, + {}, + completeCapabilities, + 'invoke_action_with_input', + 'user-a-token', + ); + const result = await raw( + ORIGINAL, + answer(first.result.requestState), + completeCapabilities, + 'invoke_action_with_input', + 'user-b-token', + ); + expect(result.error.code).toBe(-32602); + expect(execute).not.toHaveBeenCalled(); + }); + + it('allows exactly one concurrent completion and refuses later replay', async () => { + const { raw, execute } = harness(); + const first = await raw(); + const results = await Promise.all([ + raw(ORIGINAL, answer(first.result.requestState)), + raw(ORIGINAL, answer(first.result.requestState)), + ]); + expect(execute).toHaveBeenCalledOnce(); + expect(results.filter((result) => result.result.isError)).toHaveLength(1); + expect((await raw(ORIGINAL, answer(first.result.requestState))).result.isError).toBe(true); + expect(execute).toHaveBeenCalledOnce(); + }); + + it('expires pending input and bounds outstanding prompts without evicting live ones', async () => { + vi.useFakeTimers({ toFake: ['Date'] }); + const { raw, execute } = harness({ ttlSeconds: 1, maxPending: 1 }); + const first = await raw(); + expect((await raw()).result.structuredContent.diagnostics.kind).toBe('execution'); + vi.setSystemTime(Date.now() + 2_000); + expect((await raw(ORIGINAL, answer(first.result.requestState))).error.code).toBe(-32602); + expect((await raw()).result.resultType).toBe('input_required'); + expect(execute).not.toHaveBeenCalled(); + }); + + it('does not resume an unfinished prompt after closing the runtime', async () => { + const { raw, execute, completion } = harness(); + const first = await raw(); + completion.close(); + expect((await raw(ORIGINAL, answer(first.result.requestState))).error.code).toBe(-32602); + expect(execute).not.toHaveBeenCalled(); + }); + + it('lets a modern client drive the standard MRTR cycle', async () => { + const { handler, execute } = harness(); + const client = new Client( + { name: 'modern-input-test', version: '1' }, + { + capabilities: completeCapabilities, + versionNegotiation: { mode: { pin: '2026-07-28' } }, + }, + ); + client.setRequestHandler('elicitation/create', async () => ({ + action: 'accept', + content: { params_json: '{"user_id":123}' }, + })); + cleanup.push(() => client.close()); + await client.connect( + new StreamableHTTPClientTransport(new URL('http://input.invalid/mcp'), { + fetch: (input, init) => handler.fetch(new Request(input, init)), + }), + ); + const result = await client.callTool({ + name: 'invoke_action_with_input', + arguments: ORIGINAL, + }); + expect(result.structuredContent).toMatchObject({ + result: { user_id: 123, message: ORIGINAL.params.message }, + }); + expect(execute).toHaveBeenCalledOnce(); + }); + + it('does not execute when the caller aborts while the input form is open', async () => { + const { handler, execute } = harness(); + const entered = Promise.withResolvers(); + const inputAborted = Promise.withResolvers(); + const client = new Client( + { name: 'cancel-input-test', version: '1' }, + { + capabilities: completeCapabilities, + versionNegotiation: { mode: { pin: '2026-07-28' } }, + }, + ); + client.setRequestHandler('elicitation/create', async (_request, context) => { + context.mcpReq.signal.addEventListener('abort', () => inputAborted.resolve(), { + once: true, + }); + entered.resolve(); + await inputAborted.promise; + return { action: 'cancel' }; + }); + cleanup.push(() => client.close()); + await client.connect( + new StreamableHTTPClientTransport(new URL('http://input.invalid/mcp'), { + fetch: (input, init) => handler.fetch(new Request(input, init)), + }), + ); + const controller = new AbortController(); + const result = client + .callTool( + { name: 'invoke_action_with_input', arguments: ORIGINAL }, + { signal: controller.signal }, + ) + .catch((error: unknown) => error); + await entered.promise; + controller.abort(new Error('cancel test')); + await inputAborted.promise; + expect(await result).toBeInstanceOf(Error); + expect(execute).not.toHaveBeenCalled(); + }); + + it('integrates the production factory through the legacy elicitation shim', async () => { + const onebot: ActionClient = { + call: vi.fn(async () => ({ status: 'ok', retcode: 0, data: { message_id: -42 } })), + callStream: vi.fn(), + }; + vi.spyOn(console, 'error').mockImplementation(() => {}); + const server = createMcpServer({ mode: 'write', client: onebot }); + const client = new LegacyClient( + { name: 'legacy-input-test', version: '1' }, + { capabilities: { elicitation: {} } }, + ); + client.setRequestHandler(ElicitRequestSchema, async () => { + expect(onebot.call).not.toHaveBeenCalled(); + return { action: 'accept', content: { params_json: '{"user_id":123}' } }; + }); + const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair(); + cleanup.push( + () => server.close(), + () => client.close(), + ); + await server.connect(serverTransport); + await client.connect(clientTransport); + const result = await client.callTool({ + name: 'invoke_action_with_input', + arguments: ORIGINAL, + }); + expect(result.structuredContent).toMatchObject({ + result: { status: 'ok', retcode: 0, data: { message_id: -42 } }, + }); + expect(result.resultType).toBeUndefined(); + expect(onebot.call).toHaveBeenCalledExactlyOnceWith( + 'send_private_msg', + { user_id: 123, message: ORIGINAL.params.message }, + expect.objectContaining({ signal: expect.any(AbortSignal) }), + ); + }); + + it('keeps automatic local uploads complete without asking for runtime-owned transfer fields', async () => { + const streamResult = { + frame_count: 1, + frames: [], + terminal: { status: 'ok', retcode: 0, data: { file: 'uploaded-file' } }, + }; + const onebot: ActionClient = { call: vi.fn(), callStream: vi.fn(async () => streamResult) }; + vi.spyOn(console, 'error').mockImplementation(() => {}); + const server = createMcpServer({ mode: 'write', client: onebot }); + const client = new LegacyClient({ name: 'auto-upload-input-test', version: '1' }); + const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair(); + cleanup.push( + () => server.close(), + () => client.close(), + ); + await server.connect(serverTransport); + await client.connect(clientTransport); + const args = { + action: 'upload_file_stream', + execution: { input_file: '/fixture/media.bin' }, + }; + const original = await client.callTool({ name: 'invoke_action', arguments: args }); + const extended = await client.callTool({ + name: 'invoke_action_with_input', + arguments: args, + }); + expect(extended.structuredContent).toEqual(original.structuredContent); + expect(extended.isError).toBeUndefined(); + expect(onebot.callStream).toHaveBeenCalledTimes(2); + expect(onebot.callStream).toHaveBeenLastCalledWith( + 'upload_file_stream', + {}, + expect.objectContaining({ inputFile: args.execution.input_file }), + ); + }); +}); diff --git a/packages/mcp/tests/message-app.test.ts b/packages/mcp/tests/message-app.test.ts new file mode 100644 index 00000000..1d3c7133 --- /dev/null +++ b/packages/mcp/tests/message-app.test.ts @@ -0,0 +1,474 @@ +import { mkdtemp, readFile, rm } from 'node:fs/promises'; +import { createHash } from 'node:crypto'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { App } from '@modelcontextprotocol/ext-apps'; +import { AppBridge } from '@modelcontextprotocol/ext-apps/app-bridge'; +import { InMemoryTransport } from '@modelcontextprotocol/client'; +import { Client as LegacyClient } from '@modelcontextprotocol/sdk/client/index.js'; +import { createMcpHandler } from '@modelcontextprotocol/server'; +import { createMcpServer } from '../src/server-factory.js'; +import type { ActionClient } from '../src/client.js'; +import type { Mode } from '../src/tools.js'; +import { + MESSAGE_APP_MIME_TYPE, + MESSAGE_APP_URI, + messageAppResources, + messagePreviewTool, + previewMessage, + readMessageAppResource, + supportsMessageApp, +} from '../src/message-app.js'; +import { mountMessagePreview, type PreviewApp } from '../apps/preview-view.js'; +import { buildMessageApp } from '../scripts/build-message-app.mjs'; + +const temporary: string[] = []; +const close: Array<() => Promise> = []; +afterEach(async () => { + for (const cleanup of close.splice(0).reverse()) { + await cleanup(); + } + + await Promise.all( + temporary.splice(0).map((path) => rm(path, { recursive: true, force: true })), + ); +}); + +const draft = { message_type: 'group', target_id: 12345, text: '你好' }; + +function fakeOneBot(): ActionClient { + return { + call: vi.fn(async () => ({ status: 'ok', retcode: 0, data: { message_id: -42 } })), + callStream: vi.fn(async () => ({ + frame_count: 0, + frames: [], + terminal: { status: 'ok', retcode: 0 }, + })), + }; +} + +const appCapabilities = { + extensions: { 'io.modelcontextprotocol/ui': { mimeTypes: [MESSAGE_APP_MIME_TYPE] } }, +}; + +describe('message preview server integration', () => { + it.each(['docs', 'read', 'write'] as Mode[])( + 'keeps legacy preview side-effect free and applies %s mode and negotiated UI capability', + async (mode) => { + for (const ui of [false, true]) { + const onebot = fakeOneBot(); + const server = createMcpServer({ mode, client: onebot }); + const client = new LegacyClient( + { name: 'preview-legacy-test', version: '1.0.0' }, + { capabilities: ui ? appCapabilities : {} }, + ); + const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair(); + close.push( + () => server.close(), + () => client.close(), + ); + await server.connect(serverTransport); + await client.connect(clientTransport); + const tools = await client.listTools(); + const preview = tools.tools.find((tool) => tool.name === 'preview_message'); + expect(preview).toBeDefined(); + expect(preview?._meta).toEqual( + ui + ? { ui: { resourceUri: MESSAGE_APP_URI, visibility: ['model', 'app'] } } + : undefined, + ); + const resources = await client.listResources(); + expect( + resources.resources.some((resource) => resource.uri === MESSAGE_APP_URI), + ).toBe(ui); + const result = await client.callTool({ name: 'preview_message', arguments: draft }); + expect(result.isError).toBeUndefined(); + expect(result.structuredContent).toMatchObject({ + result: { can_send: mode === 'write', draft }, + }); + expect(onebot.call).not.toHaveBeenCalled(); + expect(onebot.callStream).not.toHaveBeenCalled(); + + if (mode !== 'write') { + expect( + await client.callTool({ + name: 'invoke_action', + arguments: { + action: 'send_group_msg', + params: { group_id: 12345, message: '你好' }, + }, + }), + ).toMatchObject({ isError: true }); + expect(onebot.call).not.toHaveBeenCalled(); + } + } + }, + ); + + it('negotiates UI metadata independently for each modern request and never sends a preview', async () => { + const onebot = fakeOneBot(); + const handler = createMcpHandler(() => createMcpServer({ mode: 'write', client: onebot })); + close.push(() => handler.close()); + + const request = async (method: string, params: Record, ui: boolean) => { + const response = await handler.fetch( + new Request('http://mcp.invalid/mcp', { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + Accept: 'application/json, text/event-stream', + 'Mcp-Method': method, + 'Mcp-Protocol-Version': '2026-07-28', + ...(typeof params.name === 'string' ? { 'Mcp-Name': params.name } : {}), + }, + body: JSON.stringify({ + jsonrpc: '2.0', + id: 1, + method, + params: { + ...params, + _meta: { + 'io.modelcontextprotocol/protocolVersion': '2026-07-28', + 'io.modelcontextprotocol/clientCapabilities': ui + ? appCapabilities + : {}, + 'io.modelcontextprotocol/clientInfo': { + name: 'preview-modern-test', + version: '1.0.0', + }, + }, + }, + }), + }), + ); + expect(response.status).toBe(200); + return await response.json(); + }; + + for (const ui of [true, false, true]) { + const tools = await request('tools/list', {}, ui); + expect( + tools.result.tools.find((tool: { name: string }) => tool.name === 'preview_message') + ._meta, + ).toEqual( + ui + ? { ui: { resourceUri: MESSAGE_APP_URI, visibility: ['model', 'app'] } } + : undefined, + ); + const resources = await request('resources/list', {}, ui); + expect( + resources.result.resources.some( + (resource: { uri: string }) => resource.uri === MESSAGE_APP_URI, + ), + ).toBe(ui); + expect(resources.result.cacheScope).toBe('private'); + const preview = await request( + 'tools/call', + { name: 'preview_message', arguments: draft }, + ui, + ); + expect(preview.result.structuredContent).toMatchObject({ + result: { can_send: true, draft }, + }); + } + + expect(onebot.call).not.toHaveBeenCalled(); + expect(onebot.callStream).not.toHaveBeenCalled(); + }); +}); + +describe('message preview tool', () => { + it('negotiates optional HTML support and retains text-only tool discovery', () => { + for (const capabilities of [ + undefined, + {}, + { extensions: {} }, + { extensions: { 'io.modelcontextprotocol/ui': {} } }, + ]) { + expect(supportsMessageApp(capabilities)).toBe(false); + } + + expect( + supportsMessageApp({ + extensions: { + 'io.modelcontextprotocol/ui': { mimeTypes: [MESSAGE_APP_MIME_TYPE] }, + }, + }), + ).toBe(true); + expect(messagePreviewTool(false)._meta).toBeUndefined(); + expect(messagePreviewTool(true)._meta).toEqual({ + ui: { resourceUri: MESSAGE_APP_URI, visibility: ['model', 'app'] }, + }); + expect(messageAppResources(false)).toEqual([]); + expect(messageAppResources(true)).toEqual([ + expect.objectContaining({ uri: MESSAGE_APP_URI, mimeType: MESSAGE_APP_MIME_TYPE }), + ]); + }); + + it('returns a literal text draft and preserves negative quoted-message identifiers without executing', () => { + const result = previewMessage( + { ...draft, text: '[CQ:at,qq=all]', reply_to: -42 }, + true, + ); + expect(result.isError).toBeUndefined(); + expect(result.structuredContent.result).toEqual({ + draft: { ...draft, text: '[CQ:at,qq=all]', reply_to: -42 }, + can_send: true, + notice: '此为待发送草稿,尚未发送。', + invocation: { + name: 'invoke_action', + arguments: { + action: 'send_group_msg', + params: { + group_id: 12345, + message: [ + { type: 'reply', data: { id: '-42' } }, + { + type: 'text', + data: { text: '[CQ:at,qq=all]' }, + }, + ], + }, + }, + }, + }); + expect(result.content).toEqual([ + { type: 'text', text: JSON.stringify(result.structuredContent.result, null, 2) }, + ]); + expect( + previewMessage({ ...draft, message_type: 'private' }, false).structuredContent.result, + ).toMatchObject({ + can_send: false, + invocation: { arguments: { action: 'send_private_msg', params: { user_id: 12345 } } }, + }); + }); + + it.each([ + { ...draft, message_type: 'other' }, + { ...draft, target_id: 0 }, + { ...draft, target_id: '12345' }, + { ...draft, target_id: Number.MAX_SAFE_INTEGER + 1 }, + { ...draft, text: ' \n' }, + { ...draft, reply_to: 0 }, + { ...draft, reply_to: 2147483648 }, + { ...draft, reply_to: -2147483649 }, + { ...draft, message: [] }, + ])('rejects an invalid draft without discarding fields: %j', (args) => { + expect(previewMessage(args, true)).toMatchObject({ + isError: true, + structuredContent: { diagnostics: { kind: 'invalid_arguments' } }, + }); + }); + + it('does not hide a missing packaged panel', async () => { + await expect(readMessageAppResource('unrelated://resource')).resolves.toBeUndefined(); + // Tests run from source; only an actual build supplies the sibling HTML. + await expect(readMessageAppResource(MESSAGE_APP_URI)).rejects.toMatchObject({ + code: 'ENOENT', + }); + }); + + it('builds a self-contained panel with exact CSP hashes and no external assets', async () => { + const directory = await mkdtemp(join(tmpdir(), 'snowluma-message-app-')); + temporary.push(directory); + const path = await buildMessageApp(directory); + const html = await readFile(path, 'utf8'); + const script = html.match(/