mirror of
https://github.com/morluto/rea.git
synced 2026-11-07 07:16:13 +00:00
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 <noreply@anthropic.com>
This commit is contained in:
Edgar ArakelyanandClaude Fable 5.1 committed
1 parent
18aa9af2b2
commit
46dd8f0e2b
29 files changed
+1908
-49
No files matched your search
@@ -1,8 +1,8 @@
|
||||
graph TD
|
||||
subgraph "Entry Points"
|
||||
CLI["src/cli.ts + src/cli/<br/>One-shot CLI adapter<br/>setup, diagnostics, analysis, decompilation"]
|
||||
MAIN["src/main.ts + src/main/<br/>MCP stdio server adapter<br/>config, session, process lifetime"]
|
||||
DISPATCH["scripts/rea.mjs<br/>Executable dispatcher<br/>routes mcp / one-shot"]
|
||||
MAIN["src/main.ts + src/main/<br/>MCP server adapter: stdio default, streamable HTTP opt-in<br/>config, session, process lifetime"]
|
||||
DISPATCH["scripts/rea.mjs<br/>Executable dispatcher<br/>routes mcp / mcp --http / one-shot"]
|
||||
end
|
||||
|
||||
subgraph "Production Composition"
|
||||
@@ -10,7 +10,7 @@ graph TD
|
||||
end
|
||||
|
||||
subgraph "Adapters"
|
||||
SERVER["src/server/<br/>MCP request translation<br/>createServer, tool registration, toolResult"]
|
||||
SERVER["src/server/<br/>MCP request translation<br/>createServer, tool registration, toolResult<br/>http/: streamable HTTP listener and node:http bridge"]
|
||||
HOPPER_BRIDGE["bridge/hopper_bridge.py<br/>Runs inside Hopper<br/>adapts operations to Hopper Python API"]
|
||||
GHIDRA_BRIDGE["bridge/ghidra/ReaGhidraBridge.java<br/>Packaged HeadlessScript<br/>inventory, decompiler, references, CFG"]
|
||||
INSTALL["src/cli/ + src/application/Setup*<br/>Host installation adapters<br/>macOS: verified vendor DMG to ~/Applications<br/>Linux: verified vendor package + apt/dnf/pacman"]
|
||||
|
||||
+26
@@ -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
|
||||
|
||||
+31
-1
@@ -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
|
||||
|
||||
@@ -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.
|
||||
@@ -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.
|
||||
@@ -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.
|
||||
@@ -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.
|
||||
+22
-12
@@ -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 } =
|
||||
|
||||
@@ -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<NodeJS.ProcessEnv>,
|
||||
overrides: Partial<McpHttpEndpoint> = {},
|
||||
): Result<McpHttpEndpoint, ConfigurationError> => {
|
||||
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 });
|
||||
};
|
||||
+85
-3
@@ -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<void> => {
|
||||
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,
|
||||
|
||||
@@ -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<string> = new Set(["--help", "-h"]);
|
||||
const VALUED_FLAGS: ReadonlySet<string> = 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<string, string>();
|
||||
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 }),
|
||||
};
|
||||
};
|
||||
@@ -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.";
|
||||
+35
-10
@@ -1,5 +1,11 @@
|
||||
type ShutdownSignal = "SIGINT" | "SIGTERM";
|
||||
|
||||
/** Exit status a signal-terminated process reports (128 + signal number). */
|
||||
const SIGNAL_EXIT_STATUS: Readonly<Record<ShutdownSignal, number>> = {
|
||||
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 },
|
||||
);
|
||||
@@ -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<void>;
|
||||
readonly closeOptionalProviders?: () => Promise<void>;
|
||||
readonly session: BinarySession;
|
||||
|
||||
+14
-7
@@ -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<void>;
|
||||
};
|
||||
|
||||
/** 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<void>;
|
||||
readonly closeOptionalProviders: () => Promise<void>;
|
||||
}
|
||||
@@ -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(
|
||||
|
||||
+21
-1
@@ -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<void>;
|
||||
}
|
||||
|
||||
/** 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<McpServerHandle>;
|
||||
|
||||
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;
|
||||
|
||||
@@ -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<Response>,
|
||||
): Promise<string> => {
|
||||
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<void>((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);
|
||||
});
|
||||
});
|
||||
@@ -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<string> = 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<void> => {
|
||||
target.writeHead(response.status, Object.fromEntries(response.headers));
|
||||
target.flushHeaders();
|
||||
if (response.body === null) {
|
||||
target.end();
|
||||
return;
|
||||
}
|
||||
await pipeline(Readable.fromWeb(response.body), target);
|
||||
};
|
||||
@@ -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<Parameters<typeof serveStreamableHttp>[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<string, unknown>),
|
||||
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<unknown> => {
|
||||
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();
|
||||
});
|
||||
@@ -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<string> = 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<Response>;
|
||||
}
|
||||
|
||||
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<void>;
|
||||
}
|
||||
|
||||
/** 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<boolean> => {
|
||||
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<void>;
|
||||
}
|
||||
|
||||
/** Owns the single connected server and the transport of its current session. */
|
||||
export class McpSessionSlot {
|
||||
#live: LiveSession | undefined;
|
||||
#arming: Promise<WebStandardStreamableHTTPServerTransport> | 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<Response> {
|
||||
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<void> {
|
||||
const open = this.#live;
|
||||
this.#live = undefined;
|
||||
await open?.shutdown();
|
||||
}
|
||||
|
||||
#current(): Promise<WebStandardStreamableHTTPServerTransport> {
|
||||
if (this.#live !== undefined) return Promise.resolve(this.#live.transport);
|
||||
this.#arming ??= this.#arm().finally(() => {
|
||||
this.#arming = undefined;
|
||||
});
|
||||
return this.#arming;
|
||||
}
|
||||
|
||||
async #arm(): Promise<WebStandardStreamableHTTPServerTransport> {
|
||||
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<void>,
|
||||
): Promise<void> {
|
||||
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<ReturnType<typeof createHttpServer>["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<StreamableHttpServer> => {
|
||||
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<Response> => {
|
||||
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<void> => {
|
||||
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<void> | undefined;
|
||||
const close = (): Promise<void> => {
|
||||
closing ??= (async () => {
|
||||
stopped = true;
|
||||
httpServer.closeAllConnections();
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
httpServer.close((error) =>
|
||||
error === undefined ? resolve() : reject(error),
|
||||
);
|
||||
});
|
||||
await slot.close();
|
||||
await instance.close();
|
||||
})();
|
||||
return closing;
|
||||
};
|
||||
return { url, close };
|
||||
};
|
||||
@@ -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 },
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -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<void> => {
|
||||
@@ -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)
|
||||
|
||||
@@ -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"],
|
||||
});
|
||||
},
|
||||
);
|
||||
|
||||
|
||||
@@ -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<Parameters<typeof run>[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<void> => {
|
||||
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`]);
|
||||
});
|
||||
});
|
||||
@@ -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");
|
||||
});
|
||||
});
|
||||
@@ -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<void> => {
|
||||
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,
|
||||
);
|
||||
});
|
||||
@@ -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,
|
||||
);
|
||||
});
|
||||
@@ -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<Record<string, string>>;
|
||||
readonly body?: string;
|
||||
} = {},
|
||||
): Promise<RawHttpResponse> =>
|
||||
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<RawHttpResponse> => {
|
||||
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;
|
||||
};
|
||||
@@ -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<McpHttpProcess> => {
|
||||
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<URL>((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 };
|
||||
};
|
||||
Reference in new issue
Block a user