From 46dd8f0e2b5ff74e09821e09e51bc6fd47f3bc1a Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Sat, 17 Oct 2026 21:05:10 -0700 Subject: [PATCH] feat(mcp): serve the MCP server over streamable HTTP Add rea mcp --http [--host] [--port]: the same server rea speaks over stdio, served over MCP Streamable HTTP on a loopback listener by default, using the SDK's web-standard transport already in the pinned @modelcontextprotocol/server behind a small node:http bridge. One client and one session at a time; DELETE ends the session without running provider teardown, a new client's initialize replaces a session left open, foreign Host headers are refused on loopback binds, and shutdown is signals-only. Stdio stays the default and registered transport; server.json and client registrations are unchanged. Refs morluto/rea#4933 Co-Authored-By: Claude Fable 5.1 --- docs/architecture.mermaid | 6 +- docs/cli.md | 26 ++ docs/mcp-contracts.md | 32 ++- .../agent-environments/http-transport/plan.md | 32 +++ .../agent-environments/http-transport/spec.md | 67 +++++ .../http-transport/tasks.md | 11 + .../http-transport/verification.md | 157 ++++++++++ scripts/rea.mjs | 34 ++- src/config/mcpHttpEndpoint.ts | 68 +++++ src/main.ts | 88 +++++- src/main/mcpHttpArguments.ts | 104 +++++++ src/main/messages.ts | 2 + src/main/processShutdown.ts | 45 ++- src/main/shutdown.ts | 7 +- src/main/transport.ts | 21 +- src/main/types.ts | 22 +- src/server/http/nodeHttpBridge.test.ts | 141 +++++++++ src/server/http/nodeHttpBridge.ts | 52 ++++ src/server/http/serveStreamableHttp.test.ts | 229 +++++++++++++++ src/server/http/serveStreamableHttp.ts | 272 ++++++++++++++++++ src/server/http/streamableHttpListenError.ts | 10 + src/server/shutdownRegistration.test.ts | 54 +++- tests/boundary/cli/dispatcher.test.ts | 24 +- tests/boundary/mcp/mainErrors.test.ts | 51 +++- tests/boundary/mcp/mcpHttpOptions.test.ts | 118 ++++++++ tests/boundary/mcp/streamableHttpMcp.test.ts | 85 ++++++ .../boundary/process/mcpHttpShutdown.test.ts | 39 +++ tests/support/http/rawRequest.ts | 69 +++++ tests/support/mcp/streamableHttpProcess.ts | 91 ++++++ 29 files changed, 1908 insertions(+), 49 deletions(-) create mode 100644 docs/specs/agent-environments/http-transport/plan.md create mode 100644 docs/specs/agent-environments/http-transport/spec.md create mode 100644 docs/specs/agent-environments/http-transport/tasks.md create mode 100644 docs/specs/agent-environments/http-transport/verification.md create mode 100644 src/config/mcpHttpEndpoint.ts create mode 100644 src/main/mcpHttpArguments.ts create mode 100644 src/server/http/nodeHttpBridge.test.ts create mode 100644 src/server/http/nodeHttpBridge.ts create mode 100644 src/server/http/serveStreamableHttp.test.ts create mode 100644 src/server/http/serveStreamableHttp.ts create mode 100644 src/server/http/streamableHttpListenError.ts create mode 100644 tests/boundary/mcp/mcpHttpOptions.test.ts create mode 100644 tests/boundary/mcp/streamableHttpMcp.test.ts create mode 100644 tests/boundary/process/mcpHttpShutdown.test.ts create mode 100644 tests/support/http/rawRequest.ts create mode 100644 tests/support/mcp/streamableHttpProcess.ts diff --git a/docs/architecture.mermaid b/docs/architecture.mermaid index 94b31723e0..435460319f 100644 --- a/docs/architecture.mermaid +++ b/docs/architecture.mermaid @@ -1,8 +1,8 @@ graph TD subgraph "Entry Points" CLI["src/cli.ts + src/cli/
One-shot CLI adapter
setup, diagnostics, analysis, decompilation"] - MAIN["src/main.ts + src/main/
MCP stdio server adapter
config, session, process lifetime"] - DISPATCH["scripts/rea.mjs
Executable dispatcher
routes mcp / one-shot"] + MAIN["src/main.ts + src/main/
MCP server adapter: stdio default, streamable HTTP opt-in
config, session, process lifetime"] + DISPATCH["scripts/rea.mjs
Executable dispatcher
routes mcp / mcp --http / one-shot"] end subgraph "Production Composition" @@ -10,7 +10,7 @@ graph TD end subgraph "Adapters" - SERVER["src/server/
MCP request translation
createServer, tool registration, toolResult"] + SERVER["src/server/
MCP request translation
createServer, tool registration, toolResult
http/: streamable HTTP listener and node:http bridge"] HOPPER_BRIDGE["bridge/hopper_bridge.py
Runs inside Hopper
adapts operations to Hopper Python API"] GHIDRA_BRIDGE["bridge/ghidra/ReaGhidraBridge.java
Packaged HeadlessScript
inventory, decompiler, references, CFG"] INSTALL["src/cli/ + src/application/Setup*
Host installation adapters
macOS: verified vendor DMG to ~/Applications
Linux: verified vendor package + apt/dnf/pacman"] diff --git a/docs/cli.md b/docs/cli.md index c0f8eefff4..1e93ef191d 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -310,6 +310,28 @@ Choose the guide for the target and intended interaction: Runtime requests name the target and actions. Launched targets run with your user permissions; consult the chosen guide for host requirements and effects. +## Serve MCP over streamable HTTP + +`rea mcp` speaks stdio to the client that launched it. To serve the same MCP +catalog over streamable HTTP for one local client, such as an agent environment +runtime, start the dispatcher entry with `--http`: + +```sh +rea mcp --http --port 8765 +``` + +`--host` and `--port` default to `MCP_HOST` and `MCP_PORT` from the environment, +then to `127.0.0.1` and an ephemeral port. The bound URL is written to stderr as +one JSON log line (`"msg":"MCP transport listening"`); nothing is written to +stdout. `rea mcp --http --help` lists the accepted options. The listener serves +one client at a time: a session ends with the client's `DELETE`, and a client +that connects while another session is open takes it over, so a client that +exited without ending its session needs no restart. The listener runs until +SIGINT or SIGTERM and does not stop when stdin closes. See +[transports](mcp-contracts.md#transports) for the session, `Host` header and +shutdown contract. Client registrations written by setup and the MCP registry +entry keep stdio. + ## Output and exit status `rea mcp doctor --json` checks production MCP startup and reports child logs in @@ -344,6 +366,10 @@ output formatting. | `1` | The operation could not complete. Structured output identifies invalid input, permissions, cancellation, timeouts or another failure when available. | | `128 + N` | Signal-derived status. An unhandled signal may have ended the process, with the shell or runtime preserving the conventional status; a command may instead have handled the signal, waited for its operation and cleanup to settle, and assigned the same status (`130` for SIGINT, `143` for SIGTERM). In the handled case, it may also have printed a structured `cancelled` result. | +`rea mcp --http` is long-running and prints no structured result. It exits +`130` on SIGINT or `143` on SIGTERM after closing the listener, the session and +its providers, and `1` when that cleanup fails. + A `cancelled` result does not by itself determine the exit status; it depends on the command's signal handling. Commands routed through the shared `withCommandCancellation` handler in the CLI set status `130` for SIGINT or diff --git a/docs/mcp-contracts.md b/docs/mcp-contracts.md index 7cb3a7cdc7..47cc5895b6 100644 --- a/docs/mcp-contracts.md +++ b/docs/mcp-contracts.md @@ -2,7 +2,8 @@ Use this reference to call REA tools and interpret their replies on one MCP connection. Start with [tool results](#tool-results), [retained Evidence](#retained-application-evidence-inputs), -and [session discovery](#identity-and-discovery). See [cancellation](#progress-and-cancellation) +and [session discovery](#identity-and-discovery). See [transports](#transports) +for the opt-in HTTP listener, [cancellation](#progress-and-cancellation) for partial-result recovery and [first-query deadlines](#ghidra-first-query-deadlines-and-recovery) when a cold provider exceeds the client's wait. @@ -418,6 +419,35 @@ inspection fails or is cancelled before a provider client exists. If no client remains, no provider run or health observation is available; follow the cleanup error's reported resources. +## Transports + +stdio is the default transport and the only one that client registrations, +`rea mcp doctor` and the MCP registry entry describe. `rea mcp --http [--host H] +[--port N]` is an opt-in alternative that serves the same catalog over the +streamable HTTP transport at `/mcp`; it replaces stdio for that process rather +than running beside it. Host and port come from the flags, then `MCP_HOST` and +`MCP_PORT`, then `127.0.0.1` and an ephemeral port. The bound URL is announced +as one JSON log line on stderr (`"msg":"MCP transport listening"` with `url`); +stdout stays unused. + +The HTTP server is stateful and serves one client at a time on one server +instance. `initialize` returns an `Mcp-Session-Id`; `notifications/initialized` +is acknowledged with `202`; responses arrive as JSON or SSE frames; `DELETE +/mcp` with the session id ends the session with `200`. Ending a session is not +a server shutdown: the process keeps listening, the instance keeps its providers +and the shared active target, and the next client opens a fresh session against +them. The newest client replaces an open session: an `initialize` that arrives +while a session is open ends that session instead of being refused, because a +client that exits without `DELETE` must not wedge the listener; the replaced +client's open streams are closed and its later requests are refused. Requests +that carry no valid session id are refused with `400` or `404` as the SDK +transport defines. On a loopback bind, requests whose `Host` header is outside +the SDK loopback allowlist are refused with `403`; a non-loopback bind relaxes +that check and logs the relaxation. Request bodies are bounded by the SDK's +4 MiB limit, and the response budget keeps its stdio-derived floor. The +listener ends on SIGINT or SIGTERM only, not on stdin EOF, and exits `130` or +`143` after cleanup settles. + ## Progress and cancellation REA accepts ordinary `tools/call` progress tokens. The numeric `progress` counts diff --git a/docs/specs/agent-environments/http-transport/plan.md b/docs/specs/agent-environments/http-transport/plan.md new file mode 100644 index 0000000000..c3dea3aaa1 --- /dev/null +++ b/docs/specs/agent-environments/http-transport/plan.md @@ -0,0 +1,32 @@ +# Implementation plan + +1. Widen the runtime `serve` dependency to a transport-neutral factory that may + bind asynchronously. `startMcpTransport` awaits it, hands it the server + logger, and keeps stdio as the default; the shutdown handle type loses its + stdio name. +2. Add `src/server/http/nodeHttpBridge.ts` to turn an IncomingMessage into a web + Request (headers, streamed body, abort on client close) and to pipe a web + Response back with flushed headers, and `src/server/http/serveStreamableHttp.ts` + to own one `McpServer` instance and one stateful transport per session, + ended by `DELETE` or replaced by the next client's `initialize` without + running the instance's shutdown hooks, plus plain routes, loopback Host + validation, the listening log line and idempotent close. The typed listen + error lives in its own module so the stdio entry never loads the listener. +3. Parse `--http`, `--host` and `--port` in `src/main/mcpHttpArguments.ts` from + one schema that also renders help; resolve `MCP_HOST`/`MCP_PORT` in + `src/config/mcpHttpEndpoint.ts` as a `Result` that names rejected settings + without their values. +4. Route the dispatcher to `runMcpHttpEntrypoint` when the `mcp` entry carries + `--http`; compose HTTP dependencies in `src/main.ts` with the shared + `run()` path so configuration, session, providers and shutdown stay common. +5. Register signals only for HTTP mode through `registerSignalShutdown`, a + projection of `registerProcessShutdown` without stdin that sets the signal + exit status before the idempotent shutdown runs; stdio keeps its stdin + end/close triggers. +6. Document the transport in docs/mcp-contracts.md and docs/cli.md, relabel the + architecture diagram, and cover each boundary with adapters, dispatcher, + mcp-boundary and process-boundary tests plus a Python SDK client exchange. + +No new dependency is introduced: the transport, Host validation and body bound +come from the already pinned `@modelcontextprotocol/server` root export. Tool +contracts, catalog digests, response budgets and stdio behaviour are unchanged. diff --git a/docs/specs/agent-environments/http-transport/spec.md b/docs/specs/agent-environments/http-transport/spec.md new file mode 100644 index 0000000000..95d06b67fb --- /dev/null +++ b/docs/specs/agent-environments/http-transport/spec.md @@ -0,0 +1,67 @@ +# Streamable HTTP transport for the MCP server + +Issue: https://github.com/morluto/rea/issues/4933 + +## Outcome + +`rea mcp --http [--host H] [--port N]` serves the existing MCP tool catalog over +the pinned SDK's streamable HTTP transport instead of stdio, so an agent +environment runtime or any HTTP MCP client can reach REA on a local URL. The +listener is opt-in: bare `rea mcp`, `rea mcp doctor`, `rea mcp add` and every +client registration keep stdio. + +## Requirements + +- Route `mcp --http ...` and `--mcp --http ...` to the MCP runtime in the + dispatcher without changing bare `mcp`, `mcp doctor` or `mcp add`. +- Resolve host and port from the flags, then `MCP_HOST`/`MCP_PORT` from the + environment snapshot, then `127.0.0.1` and port `0`; announce the bound URL as + one pino JSON line on stderr and leave stdout unused. +- Use `WebStandardStreamableHTTPServerTransport` in stateful mode around one + server instance built once. `DELETE` ends the session on its transport without + running the instance's shutdown hooks, so the same instance keeps its + providers and serves the next client on a fresh transport; the hooks run when + the listener closes. +- Let a new client's `initialize` replace a session that is still open, closing + the earlier transport the same way, so a client that left without `DELETE` + cannot block the listener; the replaced client's later requests are refused + (`404` while the replacing session is open). +- Bridge node:http to web Request/Response in `src/server/http/`, streaming + response chunks as they are produced and turning a client disconnect into a + request abort and body cancellation. +- Validate the `Host` header with the SDK loopback allowlist for loopback binds; + relax it for other binds and log that relaxation. +- End the HTTP listener on SIGINT/SIGTERM only, never on stdin EOF, and exit with + the signal status (`130`/`143`) after cleanup settles. +- Export `serveStreamableHttp({ host, port, createMcpServer, routes, logger })` + returning `{ url, close }` so an agent environment server can add plain routes + beside `/mcp` without a second bridge. +- Keep listener failure causes out of stderr behind a fixed message while the + typed listen error carries the system cause (such as `EADDRINUSE`) into the + error-level log line. + +## Acceptance and limits + +Adapters tests cover the bridge (method, URL, headers, body, GET without body, +incremental streaming, abort and cancel) and the listener (listening line, one +factory call across three sessions, replacement of an open session by the next +`initialize`, two `initialize` requests racing to replace one session while the +instance's shutdown hook stays installed, a working tool call after `DELETE`, +the instance's shutdown hook firing only when the listener closes, plain +routes, 404, Host refusal, port release, typed bind failure naming +`EADDRINUSE`). The dispatcher test covers the new routes and keeps `mcp add +--http` with the CLI. An mcp-boundary test spawns the built CLI and drives +initialize, tools/list, `binary_session` and `DELETE` with the SDK streamable +HTTP client, then proves `binary_session` still answers in the next session and +in a session that replaced one left open; a process-boundary test asserts +`143`/`130` exits and a released port. Concurrent clients are not supported: +the newest `initialize` owns the server. The non-loopback relaxation is covered +only through its pure allowlist helper; no wide bind runs in tests. Response +budgets keep their stdio-derived floor; registrations, `rea mcp doctor` and +`server.json` stay stdio. + +## Deferred issue criteria + +Serving an environment manifest with replayed records, its environment card and +the exported bundle are later increments of #4933. This PR must reference rather +than close the issue. diff --git a/docs/specs/agent-environments/http-transport/tasks.md b/docs/specs/agent-environments/http-transport/tasks.md new file mode 100644 index 0000000000..07981bff35 --- /dev/null +++ b/docs/specs/agent-environments/http-transport/tasks.md @@ -0,0 +1,11 @@ +# Tasks + +- [x] Define single-feature requirements and implementation plan before code. +- [x] Widen the runtime serve seam and shutdown handle to transport-neutral types. +- [x] Implement the node:http bridge and the streamable HTTP listener module. +- [x] Parse flags and `MCP_HOST`/`MCP_PORT`, route the dispatcher, add signals-only shutdown. +- [x] Document the transport, CLI entry, exit status and architecture label. +- [x] Verify bridge, listener, dispatcher, SDK client, Python client and shutdown behavior. +- [x] Keep providers alive across sessions, replace a session left open, name the bind cause in the log. +- [x] Resolve the re-review: capture the shutdown hook once, one shutdown registration primitive, flags-only HTTP routing. +- [x] Run relevant repository checks and prepare a separately scoped PR. diff --git a/docs/specs/agent-environments/http-transport/verification.md b/docs/specs/agent-environments/http-transport/verification.md new file mode 100644 index 0000000000..e261f2001b --- /dev/null +++ b/docs/specs/agent-environments/http-transport/verification.md @@ -0,0 +1,157 @@ +# Verification + +Verified base: 18aa9af2b2c26fd114113dd39528a91224801840 (fetched main). + +The worktree was moved onto that base with the slice's changes kept +uncommitted, then reinstalled with `npm ci` (exit 0; 540 packages added, 541 +audited; 6.56 s elapsed, 3.85 s user, 8.07 s sys, 881459200 bytes maximum +resident set size). After the edits of this pass the tree was rebuilt with +`npm run build` (exit 0, three turbo tasks; 5.23 s elapsed, 13.94 s user, +2.93 s sys, 2964652032 bytes) and `npm run test:prepare` (exit 0, six tasks +with three cache hits; 1.37 s elapsed, 3.07 s user, 0.76 s sys, 422625280 +bytes); `dist/mcp/` carries the rebuilt `serveStreamableHttp-*.js` chunk beside +`main.js`. + +Static gates on the final tree: `npm run check:fast` passed (typecheck and +lint, two tasks; 4.69 s elapsed, 24.25 s user, 3.00 s sys, 4480139264 bytes). +`npx oxlint` over the files touched in this pass reported no findings. After +the last edit, including this record, `npm run format` ran over 4032 files +(0.68 s elapsed, 4.11 s user, 0.74 s sys, 227393536 bytes) and `npm run +format:check` passed (0.69 s elapsed, 4.09 s user, 0.62 s sys, 216989696 +bytes). `npx knip` exited 0 (3.80 s elapsed, 5.10 s user, 0.72 s sys, +868679680 bytes). `npm run check:ci:tasks` passed all eight tasks, seven of +them turbo cache hits from the run on the same sources (0.88 s elapsed, 4.52 s +user, 1.04 s sys, 215171072 bytes): jscpd reported 2576 clones and 3.35% +duplicated lines across 3143 files under the 3.5% threshold, lint:built 0 +errors over 3734 files, and format:check ran fresh. `npm run docs:check` +passed, including the README consistency test (5 cases; 3.27 s elapsed, +3.43 s user, 0.80 s sys, 288866304 bytes). `git diff --stat` lists 13 tracked +files, none under `src/domain/`. + +Source-group tests: `/usr/bin/time -l npx vitest run --project adapters +src/server/http src/server/shutdownRegistration.test.ts` passed 3 files, 17 +tests in 2.74 s elapsed, 3.12 s user, 0.93 s sys, 289767424 bytes. +`serveStreamableHttp.test.ts` gained the concurrent case: a slot with one open +session receives two `initialize` requests through `Promise.all`, exactly one +answers `200` with a session id and the other `400`, the instance's `onclose` +hook has fired zero times, and `slot.close()` fires it once. Run before the fix +against the same file, that case failed with the hook count `0` instead of `1` +(1 failed, 5 passed; 1.07 s elapsed, 0.84 s user, 0.21 s sys, 158187520 +bytes), which is the pinned no-op hook the review described. +`shutdownRegistration.test.ts` asserts that `registerProcessShutdown` hands +the handler `SIGINT`/`SIGTERM` by name and `undefined` for stdin end, that +omitting stdin registers no input listener, and that `registerSignalShutdown` +reports `143`/`130`. + +Built-group tests ran after `npm run test:prepare`. The targeted dispatcher +run, `/usr/bin/time -l npx vitest run --project cli-boundary +tests/boundary/cli/dispatcher.test.ts`, passed 3 tests in 1.08 s elapsed, +0.89 s user, 0.26 s sys, 153026560 bytes, now including `mcp add --http` +routed to the CLI beside `mcp --http --port 0`, `--mcp --http` and `mcp --host +127.0.0.1 --http` routed to the HTTP entry and `mcp doctor --http`, `mcp add` +and `--mcp extra` keeping their routes. + +The four full projects ran in sequence while a sibling worktree ran its own +`cli-boundary`, `mcp-boundary` and `acceptance` chain on the same host. +`/usr/bin/time -l npx vitest run --project adapters` passed 247 files, 3945 +tests with 8 skipped, in 93.77 s elapsed, 80.21 s user, 30.69 s sys, 344260608 +bytes. `/usr/bin/time -l npx vitest run --project mcp-boundary` passed 155 +files with 2 skipped, 910 tests with 7 skipped, in 167.51 s elapsed, 321.24 s +user, 51.30 s sys, 4923490304 bytes (the Vitest parent, its workers and the +spawned CLI children). `/usr/bin/time -l npx vitest run --project +process-boundary` exited 1: 56 files passed, 2 skipped and 2 failed, 460 tests +passed, 19 skipped and 7 failed, in 544.81 s elapsed, 160.89 s user, 133.49 s +sys, 1028931584 bytes. `/usr/bin/time -l npx vitest run --project +cli-boundary` exited 1: 108 files passed and 3 failed, 658 tests passed, 2 +skipped and 7 failed, in 452.34 s elapsed, 734.29 s user, 234.02 s sys, +648495104 bytes. No failing file belongs to this slice. + +The failures split into two kinds. Load: six `jadxIntegration.test.ts` cases +hit the 30 s test or 10 s hook timeout, and six `javascriptApplication*.test.ts` +cases failed with the worker's result-capacity constraint; re-run alone in this +worktree after the other chains ended, `jadxIntegration.test.ts` passed all 48 +tests and both JavaScript files passed all 20 tests (`cli-boundary` three files: +20 passed, 1 failed, 23.38 s elapsed, 32.99 s user, 10.97 s sys, 343244800 +bytes; `process-boundary` two files: 48 passed, 1 failed, 62.48 s elapsed, +21.89 s user, 17.55 s sys, 686620672 bytes). Pre-existing on main in this +environment: `closedStdout.test.ts` "stops quietly when the reader closes +during streamed JSON" exits `1` instead of `141` with 1009 bytes received and +empty stderr, and `releaseArtifactCanary.test.ts` fails parsing `npm pack +--json`, which npm 12.0.2 emits as an object keyed by package name rather than +the tuple the test expects. Both reproduce identically on the clean sibling +clone at `52b73ee857029a3ab8f5c6eeef233b295951a46c` (no tracked or untracked +changes; `closedStdout` 1 failed, 3 passed, 8.86 s elapsed, 6.98 s user, 2.29 s +sys, 348012544 bytes; the canary 1 failed, 7.91 s elapsed, 7.76 s user, 1.84 s +sys, 668729344 bytes). That clone predates the verified base by 15 commits, +none touching either test or `scripts/rea.mjs`. Because the dispatcher is on +the `closedStdout` path, its scenario was also replayed against this +worktree's built `dist/` with the `18aa9af2` `scripts/rea.mjs` and with the +edited one, three runs each: all six exited `1` with 1009 bytes received and +empty stderr. Nothing in this pass was changed for these failures. + +Cross-implementation check with the Python `mcp` 1.30.0 client (httpx 0.28.1, +Python 3.12.12) against the built `node scripts/rea.mjs mcp --http --port 0 < +/dev/null`: stderr announced +`{"transport":"streamable-http","url":"http://127.0.0.1:57093/mcp","msg":"MCP transport listening"}`. +Session A: `initialize` POST returned `200 text/event-stream` with +`mcp-session-id`; `notifications/initialized` returned `202` with no body; the +standalone GET and the `tools/list` and `tools/call` POSTs returned `200 +text/event-stream`; `tools/list` reported 207 tools including +`binary_session`; `binary_session` with `{}` returned `isError=False`; `DELETE +/mcp` returned `200`. Session B, opened after that `DELETE`, initialized with a +new session id, listed the same 207 tools, and `binary_session` again returned +`isError=False` before its own `DELETE` returned `200`. The server logged two +`MCP HTTP session opened` and two `MCP HTTP session closed by the client` +lines, SIGTERM then exited it with status `143`, and stdout stayed empty. The +methods, statuses and session-id handling of session A match the recorded +reference exchange step for step; the reference's echo server labelled its +`202` and `DELETE` replies `application/json`, where this listener sends no +content type for those bodiless replies. + +## Independent review + +An adversarial review of the first implementation reproduced that, after a +`DELETE`, the next session could initialize and list tools but every tool call +failed with a `cancelled` error because the SDK's `DELETE` handling closed the +transport and the server's `onclose` hooks ran provider teardown. It also found +that a session ended without `DELETE` blocked the listener until restart, that +the bind-failure log line omitted the system cause, a duplicated foreign-`Host` +probe, a missing trailing newline on `rea mcp --http --help`, and a static +import of the HTTP listener on the stdio path. Each finding is addressed and +covered by the adapters, mcp-boundary and Python checks in this document. + +The re-review of the rebased tree raised four points, all resolved here. Two +concurrent `initialize` POSTs could leave the no-op `onclose` hook installed, +because each session end captured the hook current at that moment and the +later `finally` restored the other call's no-op; `McpSessionSlot` now captures +the instance's hook once, after the factory returns the fully wrapped server, +and every session end restores exactly that, with the adapters case above +failing before and passing after the change. The optional stdin parameter of +`registerProcessShutdown` was dead in production; `registerSignalShutdown` is +now a projection of `registerProcessShutdown(handler, { signals })`, so one +primitive owns registration while the handler learns the signal and keeps +`130`/`143`. The dispatcher routed any `mcp` invocation containing `--http` to +the HTTP entry; it now does so only when the first token after `mcp` or +`--mcp` is a flag, so `rea mcp add --http` keeps the CLI's own diagnostic, and +the dispatcher test covers that case. The formatter hunk in +`src/domain/propertyListXml.test.ts` noted in the earlier record is gone after +the rebase, as `git diff --stat` confirms. The replacement rule stays as +documented, and `docs/mcp-contracts.md` now states it in one sentence with its +rationale so maintainers can decide on it. + +## Scope and measurement limits + +The concurrent-`initialize` case is exercised at the slot with in-memory +`Request` objects, where both calls advance in lockstep; over TCP two +connections are served in separate event-loop turns, so the race is reachable +only in-process and the listener-level behaviour remains "the newest client +owns the server, the other sees `400` or `404`". The non-loopback relaxation +path is covered only through the pure allowlist helper and the loopback smoke +runs; no wide bind was exercised. Sessions are served one at a time; +concurrent clients are not supported. The two failures that reproduce on the +clean clone are environmental (macOS with Node 22.23.1 and npm 12.0.2) and are +left for their owners; the load-induced timeouts were classified by isolated +re-runs, not by repeating the full projects on an idle host. The error-level +log line that names the bind cause was observed in earlier runs, not asserted +by `mainErrors.test.ts`. Response budgets keep the stdio-derived floor and were +not re-measured over HTTP. Windows was not run. diff --git a/scripts/rea.mjs b/scripts/rea.mjs index aa0659465a..d0f5bc42a9 100755 --- a/scripts/rea.mjs +++ b/scripts/rea.mjs @@ -5,15 +5,21 @@ import { dirname, join, resolve } from "node:path"; import { fileURLToPath } from "node:url"; // Route production MCP before importing Incur. Incur owns registration helpers -// such as `mcp add`, while the MCP runtime serves the stdio tool catalog. +// such as `mcp add`, while the MCP runtime serves the tool catalog over stdio +// or, with `--http`, over streamable HTTP. const args = process.argv.slice(2); const { default: packageJson } = await import("../package.json", { with: { type: "json" }, }); process.env.REA_PACKAGE_VERSION = packageJson.version; -const isMcpMode = - args.length === 1 && (args[0] === "--mcp" || args[0] === "mcp"); +const isMcpEntry = args[0] === "--mcp" || args[0] === "mcp"; +const isMcpMode = isMcpEntry && args.length === 1; const isMcpDoctorMode = args[0] === "mcp" && args[1] === "doctor"; +// Only flags may follow the entry: `mcp add --http` stays Incur's `mcp add`. +const isMcpHttpMode = + isMcpEntry && + (args[1]?.startsWith("--") ?? false) && + args.slice(1).includes("--http"); /** Exit status of a writer terminated by SIGPIPE (128 + 13). */ const SIGPIPE_EXIT_STATUS = 141; // Set once the reader of CLI output has closed the pipe. @@ -40,15 +46,16 @@ if ( process.env.PLAYWRIGHT_BROWSERS_PATH = join(cacheRoot, "ms-playwright"); } -const runtimeFiles = isMcpMode - ? ["dist/mcp/main.js"] - : isMcpDoctorMode - ? ["dist/mcp/main.js", "dist/cli-runtime/mcpDoctor.js"] - : [ - "dist/cli-runtime/cli.js", - "dist/cli-runtime/cliOutput.js", - "dist/cli-runtime/cliResultOutput.js", - ]; +const runtimeFiles = + isMcpMode || isMcpHttpMode + ? ["dist/mcp/main.js"] + : isMcpDoctorMode + ? ["dist/mcp/main.js", "dist/cli-runtime/mcpDoctor.js"] + : [ + "dist/cli-runtime/cli.js", + "dist/cli-runtime/cliOutput.js", + "dist/cli-runtime/cliResultOutput.js", + ]; if (!(await compiledRuntimeExists(runtimeFiles))) { process.stderr.write( @@ -58,6 +65,9 @@ if (!(await compiledRuntimeExists(runtimeFiles))) { } else if (isMcpMode) { const { runEntrypoint } = await import("../dist/mcp/main.js"); await runEntrypoint(); +} else if (isMcpHttpMode) { + const { runMcpHttpEntrypoint } = await import("../dist/mcp/main.js"); + await runMcpHttpEntrypoint(args.slice(1)); } else if (isMcpDoctorMode) { guardClosedStdout(); const { runProductionMcpDoctorCli } = diff --git a/src/config/mcpHttpEndpoint.ts b/src/config/mcpHttpEndpoint.ts new file mode 100644 index 0000000000..4604497ffb --- /dev/null +++ b/src/config/mcpHttpEndpoint.ts @@ -0,0 +1,68 @@ +import { z } from "zod"; + +import { + ConfigurationError, + type ConfigurationSettingIssue, +} from "../domain/configurationErrors.js"; +import { err, ok, type Result } from "../domain/result.js"; + +/** The loopback default keeps the listener local-only until an operator opts out. */ +export const DEFAULT_MCP_HTTP_HOST = "127.0.0.1"; +/** Port 0 asks the host for an ephemeral port, reported by the listening log line. */ +export const DEFAULT_MCP_HTTP_PORT = 0; + +/** Decimal TCP port text, shared by the `--port` flag and `MCP_PORT`. */ +export const mcpHttpPortSchema = z + .string() + .regex(/^(?:0|[1-9]\d*)$/u) + .transform(Number) + .pipe(z.number().int().min(0).max(65535)); + +const mcpHttpHostSchema = z.string().trim().min(1); + +export const MCP_HTTP_PORT_CONSTRAINT = + "MCP_PORT must be a decimal integer between 0 and 65535."; +export const MCP_HTTP_HOST_CONSTRAINT = + "MCP_HOST must be a non-empty host name or IP address."; + +/** Where the streamable HTTP listener binds. */ +export interface McpHttpEndpoint { + readonly host: string; + readonly port: number; +} + +/** Resolve the endpoint: explicit values, then MCP_HOST/MCP_PORT, then loopback defaults. */ +export const parseMcpHttpEndpoint = ( + environment: Readonly, + overrides: Partial = {}, +): Result => { + const issues: ConfigurationSettingIssue[] = []; + let host = overrides.host ?? DEFAULT_MCP_HTTP_HOST; + let port = overrides.port ?? DEFAULT_MCP_HTTP_PORT; + if (overrides.host === undefined && environment.MCP_HOST !== undefined) { + const parsed = mcpHttpHostSchema.safeParse(environment.MCP_HOST); + if (parsed.success) host = parsed.data; + else + issues.push({ + setting: "MCP_HOST", + constraint: MCP_HTTP_HOST_CONSTRAINT, + }); + } + if (overrides.port === undefined && environment.MCP_PORT !== undefined) { + const parsed = mcpHttpPortSchema.safeParse(environment.MCP_PORT); + if (parsed.success) port = parsed.data; + else + issues.push({ + setting: "MCP_PORT", + constraint: MCP_HTTP_PORT_CONSTRAINT, + }); + } + if (issues.length > 0) + return err( + new ConfigurationError( + issues.map(({ constraint }) => constraint).join(" "), + { settings: issues }, + ), + ); + return ok({ host, port }); +}; diff --git a/src/main.ts b/src/main.ts index a01bd2398e..bbc36c2452 100644 --- a/src/main.ts +++ b/src/main.ts @@ -21,7 +21,14 @@ import { } from "./composition/optionalObservationProviders.js"; import { createToolResultDelivery } from "./server/toolResult.js"; import { createShutdown } from "./main/shutdown.js"; -import { registerProcessShutdown } from "./main/processShutdown.js"; +import { + registerProcessShutdown, + registerSignalShutdown, +} from "./main/processShutdown.js"; +import { + parseMcpHttpEndpoint, + type McpHttpEndpoint, +} from "./config/mcpHttpEndpoint.js"; const runtimeDependencies = (): RuntimeDependencies => ({ env: process.env, @@ -33,10 +40,36 @@ const runtimeDependencies = (): RuntimeDependencies => ({ registerShutdown: registerProcessShutdown, }); +/** Replace stdio with the streamable HTTP listener; signals alone end it, never stdin. */ +const httpRuntimeDependencies = ( + endpoint: McpHttpEndpoint, +): RuntimeDependencies => { + const base = runtimeDependencies(); + return { + ...base, + serve: async (factory, options) => { + // Loaded here so the default stdio entry never loads the HTTP listener. + const { serveStreamableHttp } = + await import("./server/http/serveStreamableHttp.js"); + return serveStreamableHttp({ + ...endpoint, + createMcpServer: factory, + ...options, + }); + }, + registerShutdown: (request) => + registerSignalShutdown((exitStatus) => { + base.setExitCode(exitStatus); + request(); + }), + }; +}; + /** * Start the long-lived MCP adapter and install idempotent shutdown handlers. - * The adapter owns its BinarySession for the process lifetime; EOF and process - * signals close bridge resources without assuming ownership of the Hopper app. + * The adapter owns its BinarySession for the process lifetime; stdin EOF on + * the stdio transport and process signals on either transport close bridge + * resources without assuming ownership of the Hopper app. */ export const run = async ( dependencies: RuntimeDependencies = runtimeDependencies(), @@ -100,6 +133,55 @@ export const runEntrypoint = async ( } }; +/** Process streams the dispatcher entries write to. */ +export interface EntrypointIo { + readonly writeStdout: (text: string) => void; + readonly writeStderr: (text: string) => void; + readonly setExitCode: (code: number) => void; +} + +const processIo: EntrypointIo = { + writeStdout: (text) => void process.stdout.write(text), + writeStderr: (text) => void process.stderr.write(text), + setExitCode: (code) => { + process.exitCode = code; + }, +}; + +/** Run `rea mcp --http`: parse its flags, resolve MCP_HOST/MCP_PORT, then serve over HTTP. */ +export const runMcpHttpEntrypoint = async ( + arguments_: readonly string[], + io: EntrypointIo = processIo, +): Promise => { + const { parseMcpHttpArguments, renderMcpHttpHelp } = + await import("./main/mcpHttpArguments.js"); + const parsed = parseMcpHttpArguments(arguments_); + if (!parsed.ok) { + if (parsed.help) { + io.writeStdout(`${renderMcpHttpHelp()}\n`); + io.setExitCode(0); + return; + } + io.writeStderr(`${parsed.message}\n`); + io.setExitCode(1); + return; + } + const endpoint = parseMcpHttpEndpoint(snapshotEnvironment(process.env), { + ...(parsed.host === undefined ? {} : { host: parsed.host }), + ...(parsed.port === undefined ? {} : { port: parsed.port }), + }); + if (!endpoint.ok) { + io.writeStderr(`${projectAnalysisError(endpoint.error).message}\n`); + io.setExitCode(1); + return; + } + await runEntrypoint( + () => run(httpRuntimeDependencies(endpoint.value)), + io.writeStderr, + io.setExitCode, + ); +}; + /** Return false when the runtime's argv path is synthetic or no longer exists. */ const isDirectEntrypoint = ( entryPath: string | undefined, diff --git a/src/main/mcpHttpArguments.ts b/src/main/mcpHttpArguments.ts new file mode 100644 index 0000000000..634c58ff32 --- /dev/null +++ b/src/main/mcpHttpArguments.ts @@ -0,0 +1,104 @@ +import { Help, z } from "incur"; + +import { + MCP_HTTP_PORT_CONSTRAINT, + mcpHttpPortSchema, +} from "../config/mcpHttpEndpoint.js"; + +/** + * Canonical `rea mcp --http` options. The help renderer lists exactly this + * schema and the parser accepts exactly these flags plus the `-h` alias. + */ +const MCP_HTTP_OPTIONS = z.object({ + http: z + .boolean() + .describe("Serve the MCP catalog over streamable HTTP at /mcp, not stdio"), + host: z + .string() + .min(1) + .optional() + .describe("Listener address; default MCP_HOST, else 127.0.0.1"), + port: z + .number() + .int() + .min(0) + .max(65535) + .optional() + .describe("Listener port; default MCP_PORT, else 0 for an ephemeral port"), + help: z.boolean().optional().describe("Show help"), +}); + +const MCP_HTTP_ALIAS = { help: "h" } as const; +const MCP_HTTP_COMMAND = "rea mcp --http"; +const MCP_HTTP_DESCRIPTION = + "Serve the MCP server over streamable HTTP for one local client"; +const HELP_FLAGS: ReadonlySet = new Set(["--help", "-h"]); +const VALUED_FLAGS: ReadonlySet = new Set(["--host", "--port"]); + +/** Render help for the dispatcher entry, which accepts no Incur global options. */ +export const renderMcpHttpHelp = (): string => + Help.formatCommand(MCP_HTTP_COMMAND, { + description: MCP_HTTP_DESCRIPTION, + hideGlobalOptions: true, + options: MCP_HTTP_OPTIONS, + alias: MCP_HTTP_ALIAS, + hint: "This entry accepts only the options above. The bound URL is logged as one JSON line on stderr.", + }); + +/** Parsed `rea mcp --http` arguments; omitted values defer to the environment. */ +export type McpHttpArguments = + | { readonly ok: true; readonly host?: string; readonly port?: number } + | { readonly ok: false; readonly help: true } + | { readonly ok: false; readonly help: false; readonly message: string }; + +const invalid = (message: string): McpHttpArguments => ({ + ok: false, + help: false, + message, +}); + +/** Split `--flag=value` into its flag and inline value. */ +const splitFlag = (argument: string): readonly [string, string | undefined] => { + const separator = argument.indexOf("="); + return separator === -1 + ? [argument, undefined] + : [argument.slice(0, separator), argument.slice(separator + 1)]; +}; + +/** Parse the dispatcher arguments after `mcp`; help wins only in flag position. */ +export const parseMcpHttpArguments = ( + arguments_: readonly string[], +): McpHttpArguments => { + const values = new Map(); + let http = false; + for (let index = 0; index < arguments_.length; index += 1) { + const argument = arguments_[index] ?? ""; + if (HELP_FLAGS.has(argument)) return { ok: false, help: true }; + if (argument === "--http") { + http = true; + continue; + } + const [flag, inline] = splitFlag(argument); + if (!VALUED_FLAGS.has(flag)) + return invalid(`Unknown mcp --http argument: ${argument}`); + const value = inline ?? arguments_[index + 1]; + if (inline === undefined) index += 1; + if (value === undefined || (inline === undefined && value.startsWith("-"))) + return invalid(`Missing value for ${flag}`); + values.set(flag, value); + } + if (!http) return invalid("--host and --port require --http"); + const host = values.get("--host")?.trim(); + if (host !== undefined && host.length === 0) + return invalid("--host must be a non-empty host name or IP address"); + const portText = values.get("--port"); + const port = + portText === undefined ? undefined : mcpHttpPortSchema.safeParse(portText); + if (port !== undefined && !port.success) + return invalid(`Invalid --port: ${MCP_HTTP_PORT_CONSTRAINT}`); + return { + ok: true, + ...(host === undefined ? {} : { host }), + ...(port === undefined ? {} : { port: port.data }), + }; +}; diff --git a/src/main/messages.ts b/src/main/messages.ts index 552be71413..965ce9f28a 100644 --- a/src/main/messages.ts +++ b/src/main/messages.ts @@ -6,3 +6,5 @@ export const MCP_SHUTDOWN_FAILED = "REA could not close cleanly. End the REA process before starting it again."; export const SERVER_START_FAILED = "REA could not start. Run `rea doctor`, then restart REA from your MCP client."; +export const MCP_HTTP_LISTEN_FAILED = + "REA could not listen for MCP HTTP connections on the selected host and port. Choose another --port or stop the process using it."; diff --git a/src/main/processShutdown.ts b/src/main/processShutdown.ts index 74c51f4d6a..1b39f90da9 100644 --- a/src/main/processShutdown.ts +++ b/src/main/processShutdown.ts @@ -1,5 +1,11 @@ type ShutdownSignal = "SIGINT" | "SIGTERM"; +/** Exit status a signal-terminated process reports (128 + signal number). */ +const SIGNAL_EXIT_STATUS: Readonly> = { + SIGINT: 130, + SIGTERM: 143, +}; + interface ShutdownSignalSource { on(event: ShutdownSignal, listener: () => void): unknown; off(event: ShutdownSignal, listener: () => void): unknown; @@ -12,7 +18,8 @@ interface ShutdownInputSource { interface ProcessShutdownEvents { readonly signals: ShutdownSignalSource; - readonly stdin: ShutdownInputSource; + /** Absent when the transport does not own stdin, as the HTTP listener does not. */ + readonly stdin?: ShutdownInputSource; } const processShutdownEvents: ProcessShutdownEvents = { @@ -20,19 +27,37 @@ const processShutdownEvents: ProcessShutdownEvents = { stdin: process.stdin, }; +/** Receives the signal that requested shutdown; stdin end and close carry none. */ +type ShutdownHandler = (signal?: ShutdownSignal) => void; + /** Register the process events that request shutdown and return their owner. */ export const registerProcessShutdown = ( - handler: () => void, + handler: ShutdownHandler, events: ProcessShutdownEvents = processShutdownEvents, ): (() => void) => { - events.signals.on("SIGINT", handler); - events.signals.on("SIGTERM", handler); - events.stdin.on("end", handler); - events.stdin.on("close", handler); + const onInterrupt = (): void => handler("SIGINT"); + const onTerminate = (): void => handler("SIGTERM"); + const onInputEnd = (): void => handler(); + events.signals.on("SIGINT", onInterrupt); + events.signals.on("SIGTERM", onTerminate); + events.stdin?.on("end", onInputEnd); + events.stdin?.on("close", onInputEnd); return () => { - events.signals.off("SIGINT", handler); - events.signals.off("SIGTERM", handler); - events.stdin.off("end", handler); - events.stdin.off("close", handler); + events.signals.off("SIGINT", onInterrupt); + events.signals.off("SIGTERM", onTerminate); + events.stdin?.off("end", onInputEnd); + events.stdin?.off("close", onInputEnd); }; }; + +/** Register signals only, for a transport that does not own stdin, and report the signal's exit status. */ +export const registerSignalShutdown = ( + handler: (exitStatus: number) => void, + signals: ShutdownSignalSource = process, +): (() => void) => + registerProcessShutdown( + (signal) => { + if (signal !== undefined) handler(SIGNAL_EXIT_STATUS[signal]); + }, + { signals }, + ); diff --git a/src/main/shutdown.ts b/src/main/shutdown.ts index 90d30c5753..c522c39829 100644 --- a/src/main/shutdown.ts +++ b/src/main/shutdown.ts @@ -1,11 +1,9 @@ -import type { StdioServerHandle } from "@modelcontextprotocol/server/stdio"; - import type { BinarySession } from "../application/binary/BinarySession.js"; import type { Logger } from "pino"; import { AnalysisError } from "../domain/analysisErrorBase.js"; import { projectAnalysisError } from "../domain/analysisErrorProjection.js"; import type { JsonValue } from "../domain/jsonValue.js"; -import type { RuntimeDependencies } from "./types.js"; +import type { McpServerHandle, RuntimeDependencies } from "./types.js"; import { MCP_SHUTDOWN_FAILED } from "./messages.js"; interface ShutdownCleanupOwner { @@ -13,8 +11,9 @@ interface ShutdownCleanupOwner { closed: boolean; } +/** Own idempotent teardown of the transport handle, providers and session for any shutdown trigger. */ export const createShutdown = (input: { - readonly handle: StdioServerHandle; + readonly handle: McpServerHandle; readonly closeAndroid?: () => Promise; readonly closeOptionalProviders?: () => Promise; readonly session: BinarySession; diff --git a/src/main/transport.ts b/src/main/transport.ts index ecc1c8e6a1..7babef962f 100644 --- a/src/main/transport.ts +++ b/src/main/transport.ts @@ -1,17 +1,18 @@ import { createAndroidAnalysisProvider } from "../composition/android.js"; import type { AndroidAnalysisPort } from "../application/android/AndroidAnalysisPort.js"; -import type { StdioServerHandle } from "@modelcontextprotocol/server/stdio"; import type { BinarySession } from "../application/binary/BinarySession.js"; import type { Logger } from "pino"; import { createServer } from "../server/createServer.js"; +import { StreamableHttpListenError } from "../server/http/streamableHttpListenError.js"; import type { ToolResultDelivery } from "../server/toolResult.js"; -import type { RuntimeDependencies } from "./types.js"; +import type { McpServerHandle, RuntimeDependencies } from "./types.js"; import type { OptionalProviderLoadResult } from "../application/OptionalObservationProviders.js"; import { projectAnalysisError } from "../domain/analysisErrorProjection.js"; import { MCP_CONNECTION_LOST, MCP_CONNECTION_START_FAILED, + MCP_HTTP_LISTEN_FAILED, } from "./messages.js"; interface ServerContext { @@ -26,6 +27,7 @@ type ClosableOptionalProvider = { readonly close?: () => Promise; }; +/** Start the injected transport (stdio or streamable HTTP) around one shared session. */ export const startMcpTransport = async ( dependencies: RuntimeDependencies, session: BinarySession, @@ -33,7 +35,7 @@ export const startMcpTransport = async ( ): Promise< | { readonly ok: true; - readonly handle: StdioServerHandle; + readonly handle: McpServerHandle; readonly closeAndroid: () => Promise; readonly closeOptionalProviders: () => Promise; } @@ -108,9 +110,9 @@ export const startMcpTransport = async ( }); await optionalProviderClose; }; - let handle: StdioServerHandle; + let handle: McpServerHandle; try { - handle = dependencies.serve( + handle = await dependencies.serve( () => { // The SDK can discard a discovery probe and construct a replacement server. const android = createAndroidAnalysisProvider( @@ -142,6 +144,7 @@ export const startMcpTransport = async ( serverLogger.error(MCP_CONNECTION_LOST); dependencies.writeStderr(`${MCP_CONNECTION_LOST}\n`); }, + logger: serverLogger, }, ); } catch (cause: unknown) { @@ -156,13 +159,17 @@ export const startMcpTransport = async ( closeServers(), closeOptionalProviders(), ]); + const startFailure = + cause instanceof StreamableHttpListenError + ? MCP_HTTP_LISTEN_FAILED + : MCP_CONNECTION_START_FAILED; serverLogger.error( { error: cause instanceof Error ? cause.message : String(cause), }, - MCP_CONNECTION_START_FAILED, + startFailure, ); - dependencies.writeStderr(`${MCP_CONNECTION_START_FAILED}\n`); + dependencies.writeStderr(`${startFailure}\n`); if (binaryCleanup.status === "fulfilled" && !binaryCleanup.value.ok) { const cleanupError = projectAnalysisError(binaryCleanup.value.error); serverLogger.error( diff --git a/src/main/types.ts b/src/main/types.ts index 8b46e25f45..d79a89390e 100644 --- a/src/main/types.ts +++ b/src/main/types.ts @@ -1,8 +1,28 @@ +import type { McpServerFactory } from "@modelcontextprotocol/server"; +import type { Logger } from "pino"; + import type { OptionalProviderLoadResult } from "../application/OptionalObservationProviders.js"; +/** Teardown handle every MCP transport returns to the runtime. */ +export interface McpServerHandle { + readonly close: () => Promise; +} + +/** Runtime inputs shared by the stdio and streamable HTTP transports. */ +export interface McpServeOptions { + readonly onerror?: (error: Error) => void; + readonly logger?: Logger; +} + +/** Start one transport around the server factory; a listener binds asynchronously. */ +export type McpServe = ( + factory: McpServerFactory, + options: McpServeOptions, +) => McpServerHandle | Promise; + export interface RuntimeDependencies { readonly env: NodeJS.ProcessEnv; - readonly serve: typeof import("@modelcontextprotocol/server/stdio").serveStdio; + readonly serve: McpServe; readonly writeStderr: (text: string) => void; readonly setExitCode: (code: number) => void; readonly registerShutdown: (handler: () => void) => () => void; diff --git a/src/server/http/nodeHttpBridge.test.ts b/src/server/http/nodeHttpBridge.test.ts new file mode 100644 index 0000000000..a906c5443d --- /dev/null +++ b/src/server/http/nodeHttpBridge.test.ts @@ -0,0 +1,141 @@ +import { once } from "node:events"; +import { createServer, type Server } from "node:http"; +import type { AddressInfo } from "node:net"; +import { afterEach, expect, it, vi } from "vitest"; + +import { toWebRequest, writeWebResponse } from "./nodeHttpBridge.js"; + +const servers: Server[] = []; +const encoder = new TextEncoder(); +const decoder = new TextDecoder(); + +afterEach(async () => { + for (const server of servers.splice(0)) { + server.closeAllConnections(); + await new Promise((resolve) => server.close(() => resolve(undefined))); + } +}); + +/** Serve a web handler through the bridge on an ephemeral loopback port. */ +const serve = async ( + handle: (request: Request) => Response | Promise, +): Promise => { + const server = createServer(); + servers.push(server); + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + const origin = `http://127.0.0.1:${String((server.address() as AddressInfo).port)}`; + server.on("request", (request, response) => { + void (async () => { + try { + const webRequest = toWebRequest(request, response, origin); + await writeWebResponse(await handle(webRequest), response); + } catch { + response.destroy(); + } + })(); + }); + return origin; +}; + +it("forwards method, URL, headers and the request body as a web Request", async () => { + const seen: Request[] = []; + const origin = await serve(async (request) => { + seen.push(request); + return Response.json( + { body: await request.text() }, + { headers: { "x-reply": "bridged" } }, + ); + }); + const response = await fetch(`${origin}/mcp?page=2`, { + method: "POST", + headers: { "content-type": "application/json", "x-probe": "yes" }, + body: '{"hello":1}', + }); + expect(response.status).toBe(200); + expect(response.headers.get("x-reply")).toBe("bridged"); + await expect(response.json()).resolves.toEqual({ body: '{"hello":1}' }); + const request = seen[0]; + if (request === undefined) throw new Error("Request not bridged"); + const url = new URL(request.url); + expect(request.method).toBe("POST"); + expect(`${url.pathname}${url.search}`).toBe("/mcp?page=2"); + expect(request.headers.get("x-probe")).toBe("yes"); + expect(request.headers.get("host")).toBe(new URL(origin).host); +}); + +it("forwards GET requests without a body", async () => { + const origin = await serve((request) => + Response.json({ bodyless: request.body === null }), + ); + const response = await fetch(`${origin}/mcp`); + await expect(response.json()).resolves.toEqual({ bodyless: true }); +}); + +it("streams response chunks as the handler produces them", async () => { + let release: () => void = () => undefined; + const gate = new Promise((resolve) => { + release = resolve; + }); + const origin = await serve( + () => + new Response( + new ReadableStream({ + async start(controller) { + controller.enqueue(encoder.encode("event: message\ndata: 1\n\n")); + await gate; + controller.enqueue(encoder.encode("event: message\ndata: 2\n\n")); + controller.close(); + }, + }), + { headers: { "content-type": "text/event-stream" } }, + ), + ); + const response = await fetch(`${origin}/stream`); + expect(response.headers.get("content-type")).toBe("text/event-stream"); + const reader = response.body?.getReader(); + if (reader === undefined) throw new Error("Response body missing"); + const first = await reader.read(); + expect(decoder.decode(first.value)).toContain("data: 1"); + release(); + let rest = ""; + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + rest += decoder.decode(value); + } + expect(rest).toContain("data: 2"); +}); + +it("aborts the web Request and cancels its response body when the client leaves", async () => { + let aborted = false; + let cancelled = false; + const origin = await serve((request) => { + request.signal.addEventListener("abort", () => { + aborted = true; + }); + return new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode(": open\n\n")); + }, + cancel() { + cancelled = true; + }, + }), + { headers: { "content-type": "text/event-stream" } }, + ); + }); + const controller = new AbortController(); + const response = await fetch(`${origin}/stream`, { + signal: controller.signal, + }); + const reader = response.body?.getReader(); + if (reader === undefined) throw new Error("Response body missing"); + await reader.read(); + controller.abort(); + await vi.waitFor(() => { + expect(aborted).toBe(true); + expect(cancelled).toBe(true); + }); +}); diff --git a/src/server/http/nodeHttpBridge.ts b/src/server/http/nodeHttpBridge.ts new file mode 100644 index 0000000000..0541021bcc --- /dev/null +++ b/src/server/http/nodeHttpBridge.ts @@ -0,0 +1,52 @@ +import type { IncomingMessage, ServerResponse } from "node:http"; +import { Readable } from "node:stream"; +import { pipeline } from "node:stream/promises"; + +/** Fetch rejects a body on these methods, so their request streams are not forwarded. */ +const BODYLESS_METHODS: ReadonlySet = new Set(["GET", "HEAD"]); + +const toHeaders = (incoming: IncomingMessage["headers"]): Headers => { + const headers = new Headers(); + for (const [name, value] of Object.entries(incoming)) { + if (value === undefined) continue; + if (Array.isArray(value)) + for (const entry of value) headers.append(name, entry); + else headers.set(name, value); + } + return headers; +}; + +/** Translate a node:http request into a web Request whose signal aborts when the client leaves. */ +export const toWebRequest = ( + request: IncomingMessage, + response: ServerResponse, + origin: string, +): Request => { + const controller = new AbortController(); + response.once("close", () => { + if (!response.writableFinished) controller.abort(); + }); + const method = request.method ?? "GET"; + return new Request(new URL(request.url ?? "/", origin), { + method, + headers: toHeaders(request.headers), + signal: controller.signal, + ...(BODYLESS_METHODS.has(method) + ? {} + : { body: Readable.toWeb(request), duplex: "half" }), + }); +}; + +/** Stream a web Response into node:http, flushing headers and each chunk as it arrives. */ +export const writeWebResponse = async ( + response: Response, + target: ServerResponse, +): Promise => { + target.writeHead(response.status, Object.fromEntries(response.headers)); + target.flushHeaders(); + if (response.body === null) { + target.end(); + return; + } + await pipeline(Readable.fromWeb(response.body), target); +}; diff --git a/src/server/http/serveStreamableHttp.test.ts b/src/server/http/serveStreamableHttp.test.ts new file mode 100644 index 0000000000..0be59585ef --- /dev/null +++ b/src/server/http/serveStreamableHttp.test.ts @@ -0,0 +1,229 @@ +import { + Client, + StreamableHTTPClientTransport, +} from "@modelcontextprotocol/client"; +import { McpServer } from "@modelcontextprotocol/server"; +import pino from "pino"; +import { afterEach, expect, it } from "vitest"; +import { z } from "zod"; + +import { + expectForeignHostRefused, + rawHttpRequest, +} from "../../../tests/support/http/rawRequest.js"; +import { + hostHeaderAllowlist, + McpSessionSlot, + serveStreamableHttp, + type StreamableHttpServer, +} from "./serveStreamableHttp.js"; +import { StreamableHttpListenError } from "./streamableHttpListenError.js"; + +const servers: StreamableHttpServer[] = []; + +afterEach(async () => { + await Promise.all(servers.splice(0).map((server) => server.close())); +}); + +/** An echo server whose protocol onclose hook counts as a shutdown, as rea's provider cleanup does. */ +const createEchoServer = (onShutdown: () => void): McpServer => { + const server = new McpServer({ name: "echo-fixture", version: "1" }); + server.registerTool( + "echo", + { description: "Echo text", inputSchema: { text: z.string() } }, + ({ text }) => ({ content: [{ type: "text", text }] }), + ); + server.server.onclose = onShutdown; + return server; +}; + +const start = async ( + overrides: Partial[0]> = {}, +) => { + const logs: string[] = []; + const logger = pino({ level: "debug" }, { write: (line) => logs.push(line) }); + let factoryCalls = 0; + let shutdownHooks = 0; + const served = await serveStreamableHttp({ + host: "127.0.0.1", + port: 0, + createMcpServer: () => { + factoryCalls += 1; + return createEchoServer(() => { + shutdownHooks += 1; + }); + }, + logger, + ...overrides, + }); + servers.push(served); + return { + served, + messages: () => + logs.map((line) => (JSON.parse(line) as { msg: string }).msg), + entries: () => + logs.map((line) => JSON.parse(line) as Record), + factoryCalls: () => factoryCalls, + shutdownHooks: () => shutdownHooks, + }; +}; + +const connect = async (url: string) => { + const transport = new StreamableHTTPClientTransport(new URL(url)); + const client = new Client({ name: "streamable-http-adapter", version: "1" }); + await client.connect(transport); + return { transport, client }; +}; + +const echo = async (client: Client, text: string): Promise => { + const result = await client.callTool({ name: "echo", arguments: { text } }); + return result.content; +}; + +/** A new client's initialize as the bridge hands it to the slot: no session id yet. */ +const initializeRequest = (): Request => + new Request("http://127.0.0.1/mcp", { + method: "POST", + headers: { + accept: "application/json, text/event-stream", + "content-type": "application/json", + }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { + protocolVersion: "2025-11-25", + capabilities: {}, + clientInfo: { name: "racing-client", version: "1" }, + }, + }), + }); + +it("announces the bound URL and serves one session at a time on one live instance", async () => { + const { served, entries, messages, factoryCalls, shutdownHooks } = + await start(); + expect(new URL(served.url).pathname).toBe("/mcp"); + expect(entries()).toContainEqual( + expect.objectContaining({ + transport: "streamable-http", + url: served.url, + msg: "MCP transport listening", + }), + ); + const first = await connect(served.url); + const sessionId = first.transport.sessionId; + expect(sessionId).toEqual(expect.any(String)); + await expect(echo(first.client, "hi")).resolves.toEqual([ + { type: "text", text: "hi" }, + ]); + // A new client's initialize replaces the open session; the earlier client is refused. + const second = await connect(served.url); + expect(second.transport.sessionId).toEqual(expect.any(String)); + expect(second.transport.sessionId).not.toBe(sessionId); + await expect(echo(first.client, "stale")).rejects.toThrow(); + await expect(echo(second.client, "live")).resolves.toEqual([ + { type: "text", text: "live" }, + ]); + await first.client.close(); + // DELETE ends the session, not the server: the instance keeps serving. + await second.transport.terminateSession(); + expect(second.transport.sessionId).toBeUndefined(); + await second.client.close(); + const third = await connect(served.url); + await expect(echo(third.client, "again")).resolves.toEqual([ + { type: "text", text: "again" }, + ]); + expect(factoryCalls()).toBe(1); + expect(shutdownHooks()).toBe(0); + expect( + messages().filter((msg) => msg === "MCP HTTP session opened"), + ).toHaveLength(3); + expect(messages()).toContain("MCP HTTP session replaced by a new client"); + // Closing the listener with a session open is the shutdown the instance's hook waits for. + await served.close(); + expect(shutdownHooks()).toBe(1); + await third.client.close(); +}); + +it("keeps the instance's shutdown hook when two initialize requests race to replace an open session", async () => { + let shutdownHooks = 0; + const slot = new McpSessionSlot( + createEchoServer(() => { + shutdownHooks += 1; + }), + undefined, + ); + const opened = await slot.handle(initializeRequest()); + expect(opened.status).toBe(200); + // Both requests see the same open session and end it in lockstep. + const raced = await Promise.all([ + slot.handle(initializeRequest()), + slot.handle(initializeRequest()), + ]); + expect(raced.map((response) => response.status).sort()).toEqual([200, 400]); + const winner = raced.find((response) => response.status === 200); + expect(winner?.headers.get("mcp-session-id")).toEqual(expect.any(String)); + expect(shutdownHooks).toBe(0); + // Only the slot's own close is the shutdown the instance's hook waits for. + await slot.close(); + expect(shutdownHooks).toBe(1); +}); + +it("serves plain routes beside the MCP endpoint and answers unknown paths with 404", async () => { + const { served } = await start({ + routes: [ + { + method: "GET", + path: "/.well-known/probe", + handle: () => Response.json({ probe: true }), + }, + ], + }); + const probe = await fetch(new URL("/.well-known/probe", served.url)); + await expect(probe.json()).resolves.toEqual({ probe: true }); + const missing = await fetch(new URL("/missing", served.url)); + expect(missing.status).toBe(404); + await expect(missing.json()).resolves.toMatchObject({ + error: { code: -32601 }, + }); + const wrongMethod = await fetch(new URL("/.well-known/probe", served.url), { + method: "POST", + body: "{}", + }); + expect(wrongMethod.status).toBe(404); +}); + +it("refuses foreign Host headers on a loopback bind and frees the port on close", async () => { + const { served } = await start(); + const url = new URL(served.url); + await expectForeignHostRefused(url); + await served.close(); + await served.close(); + await expect(rawHttpRequest(url)).rejects.toMatchObject({ + code: "ECONNREFUSED", + }); +}); + +it("reports a port that cannot be bound as a typed listen error naming the cause", async () => { + const { served } = await start(); + const failed = serveStreamableHttp({ + host: "127.0.0.1", + port: Number(new URL(served.url).port), + createMcpServer: () => createEchoServer(() => undefined), + }); + await expect(failed).rejects.toBeInstanceOf(StreamableHttpListenError); + await expect(failed).rejects.toThrow(/EADDRINUSE/u); +}); + +it("keeps the SDK loopback allowlist only for loopback binds", () => { + expect(hostHeaderAllowlist("127.0.0.1")).toEqual([ + "localhost", + "127.0.0.1", + "[::1]", + ]); + expect(hostHeaderAllowlist("localhost")).toBeDefined(); + expect(hostHeaderAllowlist("::1")).toBeDefined(); + expect(hostHeaderAllowlist("192.0.2.10")).toBeUndefined(); + expect(hostHeaderAllowlist("::")).toBeUndefined(); +}); diff --git a/src/server/http/serveStreamableHttp.ts b/src/server/http/serveStreamableHttp.ts new file mode 100644 index 0000000000..2a2fdbdc05 --- /dev/null +++ b/src/server/http/serveStreamableHttp.ts @@ -0,0 +1,272 @@ +import { + hostHeaderValidationResponse, + isInitializeRequest, + localhostAllowedHostnames, + readRequestBody, + WebStandardStreamableHTTPServerTransport, + type McpServer, + type McpServerFactory, + type Server, +} from "@modelcontextprotocol/server"; +import { randomUUID } from "node:crypto"; +import { once } from "node:events"; +import { + createServer as createHttpServer, + type IncomingMessage, + type ServerResponse, +} from "node:http"; +import type { Logger } from "pino"; + +import { toWebRequest, writeWebResponse } from "./nodeHttpBridge.js"; +import { StreamableHttpListenError } from "./streamableHttpListenError.js"; + +/** MCP endpoint path shared by `rea mcp --http` and agent environment servers. */ +export const MCP_HTTP_PATH = "/mcp"; + +/** Bind literals that keep the SDK loopback Host allowlist. */ +const LOOPBACK_HOSTS: ReadonlySet = new Set([ + "127.0.0.1", + "localhost", + "::1", + "[::1]", +]); + +/** One plain HTTP route served beside the MCP endpoint as web Request to Response. */ +export interface StreamableHttpRoute { + readonly method: string; + readonly path: string; + readonly handle: (request: Request) => Response | Promise; +} + +export interface ServeStreamableHttpOptions { + readonly host: string; + readonly port: number; + /** Called once; the instance serves every session in turn on a fresh transport. */ + readonly createMcpServer: McpServerFactory; + readonly routes?: readonly StreamableHttpRoute[]; + readonly logger?: Logger; + /** Reports listener failures after binding; request failures are logged instead. */ + readonly onerror?: (error: Error) => void; +} + +export interface StreamableHttpServer { + /** The bound MCP endpoint URL. */ + readonly url: string; + readonly close: () => Promise; +} + +/** Host header allowlist for a bind, or undefined when validation is relaxed. */ +export const hostHeaderAllowlist = (host: string): string[] | undefined => + LOOPBACK_HOSTS.has(host) ? localhostAllowedHostnames() : undefined; + +const jsonError = (status: number, code: number, message: string): Response => + Response.json( + { jsonrpc: "2.0", error: { code, message }, id: null }, + { status }, + ); + +const protocolOf = (instance: McpServer | Server): Server => + "server" in instance ? instance.server : instance; + +const describe = (cause: unknown): string => + cause instanceof Error ? cause.message : String(cause); + +const NO_SHUTDOWN_HOOK = (): void => undefined; + +/** Only a new client's initialize travels without a session id; peek within the SDK body bound. */ +const isInitializePost = async (request: Request): Promise => { + if (request.method !== "POST" || request.headers.has("mcp-session-id")) + return false; + const probe = request.clone(); + try { + const read = await readRequestBody(probe); + if (read.tooLarge) return false; + const body: unknown = JSON.parse(read.text); + return Array.isArray(body) + ? body.some((message) => isInitializeRequest(message)) + : isInitializeRequest(body); + } catch { + return false; + } finally { + await probe.body?.cancel().catch(() => undefined); + } +}; + +/** The connected transport of one session and the SDK close that ends it as a server shutdown. */ +interface LiveSession { + readonly transport: WebStandardStreamableHTTPServerTransport; + readonly shutdown: () => Promise; +} + +/** Owns the single connected server and the transport of its current session. */ +export class McpSessionSlot { + #live: LiveSession | undefined; + #arming: Promise | undefined; + /** The instance's own close hook, captured once so every session end restores exactly it. */ + readonly #shutdownHook: () => void; + + constructor( + private readonly instance: McpServer | Server, + private readonly logger: Logger | undefined, + ) { + this.#shutdownHook = protocolOf(instance).onclose ?? NO_SHUTDOWN_HOOK; + } + + /** Route one MCP request; a new client's initialize replaces a session its predecessor left open. */ + async handle(request: Request): Promise { + const open = this.#live; + if ( + open?.transport.sessionId !== undefined && + (await isInitializePost(request)) + ) { + this.logger?.info("MCP HTTP session replaced by a new client"); + await open.transport.close(); + } + return (await this.#current()).handleRequest(request); + } + + /** End the live session as part of server shutdown, so the instance's onclose runs. */ + async close(): Promise { + const open = this.#live; + this.#live = undefined; + await open?.shutdown(); + } + + #current(): Promise { + if (this.#live !== undefined) return Promise.resolve(this.#live.transport); + this.#arming ??= this.#arm().finally(() => { + this.#arming = undefined; + }); + return this.#arming; + } + + async #arm(): Promise { + const transport = new WebStandardStreamableHTTPServerTransport({ + sessionIdGenerator: () => randomUUID(), + onsessioninitialized: () => this.logger?.info("MCP HTTP session opened"), + onsessionclosed: () => + this.logger?.info("MCP HTTP session closed by the client"), + }); + const shutdown = transport.close.bind(transport); + // The SDK closes the transport on DELETE; that ends the session, not the server. + transport.close = () => this.#endSession(transport, shutdown); + await this.instance.connect(transport); + this.#live = { transport, shutdown }; + return transport; + } + + /** Close one session's transport while the instance's onclose hooks wait for close(). */ + async #endSession( + transport: WebStandardStreamableHTTPServerTransport, + shutdown: () => Promise, + ): Promise { + const protocol = protocolOf(this.instance); + protocol.onclose = NO_SHUTDOWN_HOOK; + try { + await shutdown(); + } finally { + protocol.onclose = this.#shutdownHook; + if (this.#live?.transport === transport) this.#live = undefined; + } + } +} + +const endpointUrl = ( + address: ReturnType["address"]>, +): string => { + if (address === null || typeof address === "string") + throw new TypeError("MCP HTTP listener has no TCP address"); + const host = + address.family === "IPv6" ? `[${address.address}]` : address.address; + return `http://${host}:${String(address.port)}${MCP_HTTP_PATH}`; +}; + +/** Serve one MCP server, plus plain routes, over streamable HTTP until closed. */ +export const serveStreamableHttp = async ( + options: ServeStreamableHttpOptions, +): Promise => { + const { host, port, logger, routes = [] } = options; + const allowedHosts = hostHeaderAllowlist(host); + const instance = await options.createMcpServer({ era: "legacy" }); + protocolOf(instance).onerror = (error) => + logger?.warn({ error: error.message }, "MCP HTTP request rejected"); + const slot = new McpSessionSlot(instance, logger); + let stopped = false; + const route = async (request: Request): Promise => { + if (stopped) return jsonError(503, -32000, "Server shutting down"); + const refused = + allowedHosts === undefined + ? undefined + : hostHeaderValidationResponse(request, allowedHosts); + if (refused !== undefined) return refused; + const { pathname } = new URL(request.url); + if (pathname === MCP_HTTP_PATH) return slot.handle(request); + const match = routes.find( + (candidate) => + candidate.path === pathname && candidate.method === request.method, + ); + return match === undefined + ? jsonError(404, -32601, `No route for ${request.method} ${pathname}`) + : match.handle(request); + }; + const httpServer = createHttpServer(); + try { + httpServer.listen(port, host); + await once(httpServer, "listening"); + } catch (cause: unknown) { + await instance.close().catch(() => undefined); + throw new StreamableHttpListenError(cause); + } + const url = endpointUrl(httpServer.address()); + const origin = new URL(url).origin; + const handleRequest = async ( + request: IncomingMessage, + response: ServerResponse, + ): Promise => { + try { + const webResponse = await route(toWebRequest(request, response, origin)); + await writeWebResponse(webResponse, response); + } catch (cause: unknown) { + // A client that left mid-stream is routine; a failure before headers is not. + logger?.[response.headersSent ? "debug" : "warn"]( + { error: describe(cause) }, + "MCP HTTP request did not complete", + ); + if (response.headersSent) response.destroy(); + else + await writeWebResponse( + jsonError(500, -32603, "Internal server error"), + response, + ).catch(() => undefined); + } + }; + httpServer.on("request", (request, response) => { + void handleRequest(request, response); + }); + httpServer.on("error", (error) => options.onerror?.(error)); + if (allowedHosts === undefined) + logger?.warn( + { host }, + "MCP HTTP Host header validation relaxed for a non-loopback bind", + ); + logger?.info( + { transport: "streamable-http", url }, + "MCP transport listening", + ); + let closing: Promise | undefined; + const close = (): Promise => { + closing ??= (async () => { + stopped = true; + httpServer.closeAllConnections(); + await new Promise((resolve, reject) => { + httpServer.close((error) => + error === undefined ? resolve() : reject(error), + ); + }); + await slot.close(); + await instance.close(); + })(); + return closing; + }; + return { url, close }; +}; diff --git a/src/server/http/streamableHttpListenError.ts b/src/server/http/streamableHttpListenError.ts new file mode 100644 index 0000000000..7f536f9207 --- /dev/null +++ b/src/server/http/streamableHttpListenError.ts @@ -0,0 +1,10 @@ +/** The listener could not bind before any client connected; the message names the system cause. */ +export class StreamableHttpListenError extends Error { + override readonly name = "StreamableHttpListenError"; + constructor(cause: unknown) { + super( + `MCP HTTP listener could not bind: ${cause instanceof Error ? cause.message : String(cause)}`, + { cause }, + ); + } +} diff --git a/src/server/shutdownRegistration.test.ts b/src/server/shutdownRegistration.test.ts index 70da2d2dcc..41eb94ecec 100644 --- a/src/server/shutdownRegistration.test.ts +++ b/src/server/shutdownRegistration.test.ts @@ -3,11 +3,55 @@ import { expect, it } from "vitest"; import pino from "pino"; import { createShutdown } from "../main/shutdown.js"; -import { registerProcessShutdown } from "../main/processShutdown.js"; -import type { RuntimeDependencies } from "../main/types.js"; -import type { StdioServerHandle } from "@modelcontextprotocol/server/stdio"; +import { + registerProcessShutdown, + registerSignalShutdown, +} from "../main/processShutdown.js"; +import type { McpServerHandle, RuntimeDependencies } from "../main/types.js"; import { createManagedBinarySession } from "../composition/binary.js"; +it("registers only signals for a listener and reports the signal exit status", () => { + const signals = new EventEmitter(); + const statuses: number[] = []; + const unregister = registerSignalShutdown( + (status) => statuses.push(status), + signals, + ); + signals.emit("SIGTERM"); + signals.emit("SIGINT"); + expect(statuses).toEqual([143, 130]); + unregister(); + signals.emit("SIGTERM"); + expect(statuses).toEqual([143, 130]); + expect(signals.listenerCount("SIGTERM")).toBe(0); +}); + +it("names the signal that fired, passes none for stdin, and ignores stdin when the events omit it", () => { + const signals = new EventEmitter(); + const input = new EventEmitter(); + const triggers: (string | undefined)[] = []; + const withInput = registerProcessShutdown((signal) => triggers.push(signal), { + signals, + stdin: input, + }); + signals.emit("SIGINT"); + input.emit("end"); + expect(triggers).toEqual(["SIGINT", undefined]); + withInput(); + const signalsOnly = registerProcessShutdown( + (signal) => triggers.push(signal), + { signals }, + ); + signals.emit("SIGTERM"); + input.emit("end"); + input.emit("close"); + expect(triggers).toEqual(["SIGINT", undefined, "SIGTERM"]); + signalsOnly(); + signals.emit("SIGINT"); + expect(triggers).toEqual(["SIGINT", undefined, "SIGTERM"]); + expect(input.listenerCount("end")).toBe(0); +}); + const shutdownEvents = ["SIGINT", "SIGTERM", "end", "close"] as const; const flush = async (): Promise => { @@ -51,7 +95,7 @@ it.each(shutdownEvents)( }; }, }; - const handle: StdioServerHandle = { + const handle: McpServerHandle = { close: async () => { transportCloseCalls += 1; if (transportCloseCalls === 1) @@ -134,7 +178,7 @@ it("joins concurrent shutdown events and retries only failed owners", async () = registerShutdown: (handler) => registerProcessShutdown(handler, { signals, stdin: input }), }; - const handle: StdioServerHandle = { + const handle: McpServerHandle = { close: async () => { transportCloseCalls += 1; if (transportCloseCalls === 1) diff --git a/tests/boundary/cli/dispatcher.test.ts b/tests/boundary/cli/dispatcher.test.ts index 516d33b68f..33fab705af 100644 --- a/tests/boundary/cli/dispatcher.test.ts +++ b/tests/boundary/cli/dispatcher.test.ts @@ -56,7 +56,7 @@ describe("executable dispatcher", () => { ), workspace.write( "dist/mcp/main.js", - 'export const runEntrypoint = () => process.stdout.write(JSON.stringify({ adapter: "mcp" }));\n', + 'export const runEntrypoint = () => process.stdout.write(JSON.stringify({ adapter: "mcp" })); export const runMcpHttpEntrypoint = (args) => process.stdout.write(JSON.stringify({ adapter: "mcp-http", args }));\n', ), workspace.write( "dist/cli-runtime/cli.js", @@ -100,6 +100,28 @@ describe("executable dispatcher", () => { adapter: "cli", args: ["--mcp", "extra"], }); + await expect(invoke(["mcp", "--http", "--port", "0"])).resolves.toEqual({ + adapter: "mcp-http", + args: ["--http", "--port", "0"], + }); + await expect(invoke(["--mcp", "--http"])).resolves.toEqual({ + adapter: "mcp-http", + args: ["--http"], + }); + await expect( + invoke(["mcp", "--host", "127.0.0.1", "--http"]), + ).resolves.toEqual({ + adapter: "mcp-http", + args: ["--host", "127.0.0.1", "--http"], + }); + await expect(invoke(["mcp", "doctor", "--http"])).resolves.toEqual({ + adapter: "mcp-doctor", + args: ["--http"], + }); + await expect(invoke(["mcp", "add", "--http"])).resolves.toEqual({ + adapter: "cli", + args: ["mcp", "add", "--http"], + }); }, ); diff --git a/tests/boundary/mcp/mainErrors.test.ts b/tests/boundary/mcp/mainErrors.test.ts index c3738e5e2d..f20512653f 100644 --- a/tests/boundary/mcp/mainErrors.test.ts +++ b/tests/boundary/mcp/mainErrors.test.ts @@ -1,10 +1,12 @@ -import type { ServeStdioOptions } from "@modelcontextprotocol/server/stdio"; import { spawnSync } from "node:child_process"; import { resolve } from "node:path"; import { pathToFileURL } from "node:url"; import { describe, expect, it } from "vitest"; import { run, runEntrypoint } from "../../../src/main.js"; +import { MCP_HTTP_LISTEN_FAILED } from "../../../src/main/messages.js"; +import type { McpServeOptions } from "../../../src/main/types.js"; +import { StreamableHttpListenError } from "../../../src/server/http/streamableHttpListenError.js"; type RuntimeDependencies = NonNullable[0]>; @@ -61,7 +63,7 @@ describe("MCP runtime errors", () => { const output: string[] = []; const shutdown: Array<() => void> = []; const shutdownUnregistrations: string[] = []; - let options: ServeStdioOptions | undefined; + let options: McpServeOptions | undefined; let closeCalls = 0; const close = (): Promise => { closeCalls += 1; @@ -176,3 +178,48 @@ describe("MCP runtime errors", () => { expect(exitCodes).toEqual([1]); }); }); + +describe("MCP runtime HTTP transport errors", () => { + it("awaits an asynchronous transport start before installing shutdown", async () => { + const shutdown: Array<() => void> = []; + let closeCalls = 0; + expect( + await run( + dependencies({ + serve: async () => ({ + close: async () => { + closeCalls += 1; + }, + }), + output: [], + shutdown, + exitCodes: [], + }), + ), + ).toBe(0); + expect(shutdown).toHaveLength(1); + shutdown[0]?.(); + await nextTurn(); + expect(closeCalls).toBe(1); + }); + + it("reports an HTTP listener that cannot bind without its cause", async () => { + const output: string[] = []; + expect( + await run( + dependencies({ + serve: () => + Promise.reject( + new StreamableHttpListenError( + new Error("SECRET EADDRINUSE 127.0.0.1:1"), + ), + ), + output, + shutdown: [], + exitCodes: [], + }), + ), + ).toBe(1); + expect(output).toEqual([`${MCP_HTTP_LISTEN_FAILED}\n`]); + }); +}); diff --git a/tests/boundary/mcp/mcpHttpOptions.test.ts b/tests/boundary/mcp/mcpHttpOptions.test.ts new file mode 100644 index 0000000000..243ef5682d --- /dev/null +++ b/tests/boundary/mcp/mcpHttpOptions.test.ts @@ -0,0 +1,118 @@ +import { describe, expect, it } from "vitest"; + +import { parseMcpHttpEndpoint } from "../../../src/config/mcpHttpEndpoint.js"; +import { projectAnalysisError } from "../../../src/domain/analysisErrorProjection.js"; +import { runMcpHttpEntrypoint } from "../../../src/main.js"; +import { + parseMcpHttpArguments, + renderMcpHttpHelp, +} from "../../../src/main/mcpHttpArguments.js"; + +describe("rea mcp --http arguments", () => { + it("accepts --http with optional host and port in either flag spelling", () => { + expect(parseMcpHttpArguments(["--http"])).toEqual({ ok: true }); + expect( + parseMcpHttpArguments(["--http", "--host", "::1", "--port=8080"]), + ).toEqual({ ok: true, host: "::1", port: 8080 }); + expect(parseMcpHttpArguments(["--port", "0", "--http"])).toEqual({ + ok: true, + port: 0, + }); + }); + + it("answers help only in flag position and names the rejected argument", () => { + expect(parseMcpHttpArguments(["--http", "-h"])).toEqual({ + ok: false, + help: true, + }); + expect(parseMcpHttpArguments(["--http", "--host", "--help"])).toEqual({ + ok: false, + help: false, + message: "Missing value for --host", + }); + expect(parseMcpHttpArguments(["--http", "add"])).toMatchObject({ + ok: false, + message: "Unknown mcp --http argument: add", + }); + expect(parseMcpHttpArguments(["--port", "1"])).toMatchObject({ + ok: false, + message: "--host and --port require --http", + }); + for (const port of ["70000", "08", "-1", "http"]) + expect(parseMcpHttpArguments(["--http", `--port=${port}`])).toMatchObject( + { ok: false, message: expect.stringContaining("Invalid --port") }, + ); + expect(parseMcpHttpArguments(["--http", "--host", " "])).toMatchObject({ + ok: false, + message: expect.stringContaining("--host must be"), + }); + }); + + it("renders help from the option schema without Incur global options", () => { + const help = renderMcpHttpHelp(); + expect(help).toContain("rea mcp --http"); + expect(help).toContain("--host"); + expect(help).toContain("--port"); + expect(help).toContain("MCP_PORT"); + expect(help).not.toContain("--filter-output"); + }); + + it("writes help as one newline-terminated stdout block and rejections to stderr", async () => { + const io = () => { + const stdout: string[] = []; + const stderr: string[] = []; + const exitCodes: number[] = []; + return { + stdout, + stderr, + exitCodes, + writeStdout: (text: string) => void stdout.push(text), + writeStderr: (text: string) => void stderr.push(text), + setExitCode: (code: number) => void exitCodes.push(code), + }; + }; + const help = io(); + await runMcpHttpEntrypoint(["--http", "--help"], help); + expect(help.stdout).toEqual([`${renderMcpHttpHelp()}\n`]); + expect(help.stderr).toEqual([]); + expect(help.exitCodes).toEqual([0]); + const rejected = io(); + await runMcpHttpEntrypoint(["--http", "add"], rejected); + expect(rejected.stdout).toEqual([]); + expect(rejected.stderr).toEqual(["Unknown mcp --http argument: add\n"]); + expect(rejected.exitCodes).toEqual([1]); + }); +}); + +describe("MCP_HOST and MCP_PORT", () => { + it("defaults to loopback and an ephemeral port", () => { + expect(parseMcpHttpEndpoint({})).toEqual({ + ok: true, + value: { host: "127.0.0.1", port: 0 }, + }); + }); + + it("reads the environment snapshot and lets explicit values win", () => { + expect( + parseMcpHttpEndpoint({ MCP_HOST: "192.0.2.10", MCP_PORT: "8080" }), + ).toEqual({ ok: true, value: { host: "192.0.2.10", port: 8080 } }); + expect( + parseMcpHttpEndpoint( + { MCP_HOST: " ", MCP_PORT: "nope" }, + { host: "::1", port: 1 }, + ), + ).toEqual({ ok: true, value: { host: "::1", port: 1 } }); + }); + + it("rejects malformed values by setting name without echoing them", () => { + const result = parseMcpHttpEndpoint({ MCP_HOST: " ", MCP_PORT: "65536" }); + if (result.ok) throw new Error("Expected a configuration error"); + expect(result.error.settings.map(({ setting }) => setting)).toEqual([ + "MCP_HOST", + "MCP_PORT", + ]); + const projected = JSON.stringify(projectAnalysisError(result.error)); + expect(projected).toContain("MCP_PORT must be a decimal integer"); + expect(projected).not.toContain("65536"); + }); +}); diff --git a/tests/boundary/mcp/streamableHttpMcp.test.ts b/tests/boundary/mcp/streamableHttpMcp.test.ts new file mode 100644 index 0000000000..64560b20d2 --- /dev/null +++ b/tests/boundary/mcp/streamableHttpMcp.test.ts @@ -0,0 +1,85 @@ +import { + Client, + StreamableHTTPClientTransport, +} from "@modelcontextprotocol/client"; +import { describe, expect } from "vitest"; + +import { expectForeignHostRefused } from "../../support/http/rawRequest.js"; +import { startMcpHttpProcess } from "../../support/mcp/streamableHttpProcess.js"; +import { processTest } from "../../support/process/processFixture.js"; +import { assertAdvertisedMcpContracts } from "./mcpContractHarness.js"; + +const HTTP_BOUNDARY_TIMEOUT_MS = 90_000; + +const openSession = async (url: URL, name: string) => { + const transport = new StreamableHTTPClientTransport(url); + const client = new Client({ name, version: "1" }); + await client.connect(transport); + expect(transport.sessionId).toEqual(expect.any(String)); + return { transport, client }; +}; + +/** The shared target answers with its identity only while its providers are still open. */ +const expectIdentity = async (client: Client): Promise => { + const identity = await client.callTool({ + name: "binary_session", + arguments: {}, + }); + expect(identity.isError).toBeFalsy(); + expect(identity.structuredContent).toMatchObject({ + result: { + server_identity: { package: { version: expect.any(String) } }, + }, + }); +}; + +describe("rea mcp --http", () => { + processTest( + "serves initialize, tools/list, a tool call and DELETE, then keeps serving later clients", + async ({ processes }) => { + const started = await startMcpHttpProcess(processes); + expect(started.url.hostname).toBe("127.0.0.1"); + expect(started.url.pathname).toBe("/mcp"); + let stdout = ""; + started.child.stdout.on("data", (chunk: Buffer) => { + stdout += chunk.toString("utf8"); + }); + const first = await openSession(started.url, "streamable-http-boundary"); + const { tools } = await first.client.listTools(); + assertAdvertisedMcpContracts(["binary_session"], tools); + await expectIdentity(first.client); + await first.transport.terminateSession(); + expect(first.transport.sessionId).toBeUndefined(); + await first.client.close(); + // DELETE ended the session, not the server: the next client's tools still work. + const second = await openSession(started.url, "next-session"); + await expectIdentity(second.client); + // A client that leaves without DELETE does not block the listener: the next initialize replaces it. + const third = await openSession(started.url, "replacing-session"); + expect(third.transport.sessionId).not.toBe(second.transport.sessionId); + await expectIdentity(third.client); + await expect( + second.client.callTool({ name: "binary_session", arguments: {} }), + ).rejects.toThrow(); + await second.client.close(); + await third.transport.terminateSession(); + await third.client.close(); + expect(stdout).toBe(""); + expect(started.stderr()).toContain('"transport":"streamable-http"'); + expect(started.stderr()).toContain( + "MCP HTTP session replaced by a new client", + ); + }, + HTTP_BOUNDARY_TIMEOUT_MS, + ); + + processTest( + "refuses a non-loopback Host header on the loopback bind", + async ({ processes }) => { + const { url } = await startMcpHttpProcess(processes); + const refused = await expectForeignHostRefused(url); + expect(refused.body).not.toContain("127.0.0.1"); + }, + HTTP_BOUNDARY_TIMEOUT_MS, + ); +}); diff --git a/tests/boundary/process/mcpHttpShutdown.test.ts b/tests/boundary/process/mcpHttpShutdown.test.ts new file mode 100644 index 0000000000..6b1ffdbde3 --- /dev/null +++ b/tests/boundary/process/mcpHttpShutdown.test.ts @@ -0,0 +1,39 @@ +import { once } from "node:events"; +import { describe, expect } from "vitest"; + +import { MCP_SHUTDOWN_FAILED } from "../../../src/main/messages.js"; +import { rawHttpRequest } from "../../support/http/rawRequest.js"; +import { startMcpHttpProcess } from "../../support/mcp/streamableHttpProcess.js"; +import { processTest } from "../../support/process/processFixture.js"; + +const SHUTDOWN_TIMEOUT_MS = 90_000; + +const shutdownSignals: readonly (readonly [NodeJS.Signals, number])[] = + process.platform === "win32" + ? [["SIGINT", 130]] + : [ + ["SIGTERM", 143], + ["SIGINT", 130], + ]; + +describe("rea mcp --http shutdown", () => { + for (const [signal, status] of shutdownSignals) + processTest( + `releases the listener and exits ${String(status)} on ${signal}`, + async ({ processes }) => { + const started = await startMcpHttpProcess(processes); + // The live transport refuses a GET that does not accept an event stream. + const before = await rawHttpRequest(started.url); + expect(before.status).toBe(406); + started.child.kill(signal); + const [code, signalCode] = await once(started.child, "close"); + expect(code).toBe(status); + expect(signalCode).toBeNull(); + expect(started.stderr()).not.toContain(MCP_SHUTDOWN_FAILED); + await expect(rawHttpRequest(started.url)).rejects.toMatchObject({ + code: "ECONNREFUSED", + }); + }, + SHUTDOWN_TIMEOUT_MS, + ); +}); diff --git a/tests/support/http/rawRequest.ts b/tests/support/http/rawRequest.ts new file mode 100644 index 0000000000..8f87c26547 --- /dev/null +++ b/tests/support/http/rawRequest.ts @@ -0,0 +1,69 @@ +import { request as httpRequest, type IncomingHttpHeaders } from "node:http"; + +import { expect } from "vitest"; + +/** One buffered HTTP response. */ +export interface RawHttpResponse { + readonly status: number; + readonly headers: IncomingHttpHeaders; + readonly body: string; +} + +/** Send one request with caller-controlled headers, including Host, which fetch overwrites. */ +export const rawHttpRequest = ( + url: URL, + options: { + readonly method?: string; + readonly headers?: Readonly>; + readonly body?: string; + } = {}, +): Promise => + new Promise((resolve, reject) => { + // A fresh connection per request keeps a closed listener observable as ECONNREFUSED. + const request = httpRequest( + url, + { + method: options.method ?? "GET", + headers: { ...options.headers }, + agent: false, + }, + (response) => { + let body = ""; + response.setEncoding("utf8"); + response.on("data", (chunk: string) => { + body += chunk; + }); + response.on("end", () => + resolve({ + status: response.statusCode ?? 0, + headers: response.headers, + body, + }), + ); + response.on("error", reject); + }, + ); + request.on("error", reject); + request.end(options.body); + }); + +/** Assert the MCP endpoint refuses a Host header outside the loopback allowlist. */ +export const expectForeignHostRefused = async ( + url: URL, +): Promise => { + const refused = await rawHttpRequest(url, { + method: "POST", + headers: { + host: "rea.invalid", + "content-type": "application/json", + accept: "application/json, text/event-stream", + }, + body: "{}", + }); + expect(refused.status).toBe(403); + expect(JSON.parse(refused.body)).toMatchObject({ + jsonrpc: "2.0", + error: { code: -32000 }, + }); + return refused; +}; diff --git a/tests/support/mcp/streamableHttpProcess.ts b/tests/support/mcp/streamableHttpProcess.ts new file mode 100644 index 0000000000..35d8ff8945 --- /dev/null +++ b/tests/support/mcp/streamableHttpProcess.ts @@ -0,0 +1,91 @@ +import type { ChildProcessWithoutNullStreams } from "node:child_process"; +import { resolve as resolvePath } from "node:path"; + +import type { TestProcesses } from "../process/processFixture.js"; + +const LISTENING_MESSAGE = "MCP transport listening"; +const STARTUP_TIMEOUT_MS = 30_000; + +/** A compiled `rea mcp --http` child and the URL it announced. */ +export interface McpHttpProcess { + readonly child: ChildProcessWithoutNullStreams; + readonly url: URL; + /** Everything the child has written to stderr so far. */ + readonly stderr: () => string; +} + +/** Find the bound URL in the pino JSON lines the listener writes to stderr. */ +const listeningUrl = (stderr: string): URL | undefined => { + for (const line of stderr.split("\n")) { + let entry: unknown; + try { + entry = JSON.parse(line); + } catch { + continue; + } + if ( + typeof entry === "object" && + entry !== null && + "msg" in entry && + entry.msg === LISTENING_MESSAGE && + "url" in entry && + typeof entry.url === "string" + ) + return new URL(entry.url); + } + return undefined; +}; + +/** Spawn `rea mcp --http --port 0` with stdin closed and resolve once it names its URL. */ +export const startMcpHttpProcess = async ( + processes: TestProcesses, + arguments_: readonly string[] = [], +): Promise => { + const child = processes.spawn( + process.execPath, + [ + resolvePath("scripts/rea.mjs"), + "mcp", + "--http", + "--port", + "0", + ...arguments_, + ], + { + env: { + PATH: process.env.PATH ?? "", + HOPPER_LAUNCHER_PATH: "/rea-unconfigured-deep-provider/hopper", + }, + }, + ); + // A closed stdin must not end the HTTP listener; only signals do. + child.stdin.end(); + let stderr = ""; + const url = await new Promise((resolve, reject) => { + const timer = setTimeout( + () => + reject( + new Error( + `No listening line within ${String(STARTUP_TIMEOUT_MS)} ms:\n${stderr}`, + ), + ), + STARTUP_TIMEOUT_MS, + ); + child.stderr.on("data", (chunk: Buffer) => { + stderr += chunk.toString("utf8"); + const found = listeningUrl(stderr); + if (found === undefined) return; + clearTimeout(timer); + resolve(found); + }); + child.once("exit", (code, signal) => { + clearTimeout(timer); + reject( + new Error( + `Listener exited early (${String(code)}, ${String(signal)}):\n${stderr}`, + ), + ); + }); + }); + return { child, url, stderr: () => stderr }; +};