diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 000000000..186ff3d12 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,10 @@ +node_modules +.git +test +specs +deploy +website +data +dist +.claude +.specify diff --git a/.env.example b/.env.example index 77ca0f3a3..46c7492e9 100644 --- a/.env.example +++ b/.env.example @@ -149,6 +149,10 @@ # III_STREAMS_PORT=3112 # Streams API port # III_ENGINE_URL=ws://localhost:49134 # iii-engine WebSocket URL (used by the worker) +# AGENTMEMORY_MCP_PORT=3114 # MCP Streamable HTTP port (JSON-RPC transport). Defaults to REST port + 3. +# # MCP clients that speak Streamable HTTP can connect directly to +# # http://localhost:3114/mcp instead of spawning npx @agentmemory/mcp. + # ----------------------------------------------------------------------------- # 8. iii engine pin # ----------------------------------------------------------------------------- diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 000000000..b5e36c3f6 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,27 @@ +ARG III_VERSION=0.11.2 + +FROM node:22-slim AS build +WORKDIR /app +COPY package.json package-lock.json tsconfig.json tsdown.config.ts ./ +RUN npm ci +COPY src/ ./src/ +RUN npm run build + +FROM iiidev/iii:${III_VERSION} AS iii-image + +FROM node:22-slim AS runtime +RUN apt-get update \ + && apt-get install -y --no-install-recommends ca-certificates tini curl \ + && rm -rf /var/lib/apt/lists/* +ARG III_VERSION +COPY --from=iii-image /app/iii /usr/local/bin/iii +WORKDIR /app +COPY --from=build /app/dist/ ./dist/ +COPY --from=build /app/package.json /app/package-lock.json ./ +RUN npm ci --omit=dev --no-fund --no-audit +# Fallback default config; entrypoint.sh overwrites this for Docker at startup +COPY iii-config.docker.yaml ./iii-config.yaml +COPY entrypoint.sh ./ +RUN chmod +x entrypoint.sh +EXPOSE 3111 3112 3113 3114 +ENTRYPOINT ["/usr/bin/tini", "--", "/app/entrypoint.sh"] diff --git a/README.md b/README.md index c4ec2c1e0..6d794eb84 100644 --- a/README.md +++ b/README.md @@ -498,6 +498,49 @@ Imported sessions show up in the Replay picker alongside native ones. Under the > **Heads-up if you rely on `import-jsonl` as your primary capture path:** Claude Code's `cleanupPeriodDays` (in `~/.claude/settings.json`, default **30**) auto-deletes JSONL transcripts older than that window from `~/.claude/projects/`. If you install agentmemory fresh on a months-old Claude Code history, anything older than 30 days is already gone before the first import. Either run `import-jsonl` on a cron, raise `cleanupPeriodDays` to something higher, or wire the auto-capture hooks (the default plugin install path) so each turn lands in agentmemory while the session is live and the JSONL cleanup stops mattering. +### Docker + +Run agentmemory in a container with one command — no Node.js or npm required on the host. + +**Prerequisites:** Docker and docker-compose. + +```bash +# Clone and start — iii engine + agentmemory worker + viewer + MCP Stream HTTP +git clone https://github.com/rohitg00/agentmemory.git +cd agentmemory +docker compose build && docker compose up -d + +# Grab the auto-generated HMAC secret for MCP client auth +docker compose logs agentmemory | grep AGENTMEMORY_SECRET +# or: docker compose exec agentmemory cat /data/.hmac + +# Verify it is up +curl -s http://localhost:3111/agentmemory/health | head -c 200 +``` + +MCP clients that speak Streamable HTTP can connect directly without spawning +an `npx` process: + +```json +{ + "mcpServers": { + "agentmemory": { + "url": "http://localhost:3114/mcp", + "headers": { + "Authorization": "Bearer " + } + } + } +} +``` + +Stdio-based MCP clients (Cursor, Claude Code, etc.) still work through the +`npx @agentmemory/mcp` shim as usual — set `AGENTMEMORY_URL=http://localhost:3111` +and `AGENTMEMORY_SECRET` in the client's env block. + +See [specs/001-mcp-stream-dockerize/quickstart.md](specs/001-mcp-stream-dockerize/quickstart.md) +for advanced Docker setups (custom ports, external volumes, multi-instance). + ### Upgrade / Maintenance Use the maintenance command when you intentionally want to update your local runtime: diff --git a/docker-compose.yml b/docker-compose.yml index 6b1176521..e8ab787c8 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,10 +1,4 @@ services: - # One-shot init container: docker creates named volumes root:root mode - # 755, but the iii-engine image is distroless and runs as UID 65532 - # with no `chown` of its own. Without this, /data is unwritable, the - # engine silently buffers in RAM, and state evaporates on every - # restart — the exact symptom v0.9.7's working-directory fix set out - # to solve. Runs once at compose-up and exits. iii-init: image: busybox:1.36 user: "0:0" @@ -14,15 +8,6 @@ services: restart: "no" iii-engine: - # Pinned to v0.11.2 — the last engine that runs agentmemory's current - # worker model cleanly. v0.11.6 introduces a new sandbox-everything- - # via-`iii worker add` model that agentmemory hasn't been refactored - # for yet; the architectural mismatch surfaces as EPIPE reconnect - # loops and empty search after save. Bump only after agentmemory is - # refactored to register as a sandboxed worker. - # - # Override per-shell or via .env file: - # AGENTMEMORY_III_VERSION=0.11.7 docker compose up image: iiidev/iii:${AGENTMEMORY_III_VERSION:-0.11.2} user: "65532:65532" depends_on: @@ -36,6 +21,45 @@ services: volumes: - iii-data:/data - ./iii-config.docker.yaml:/app/config.yaml:ro + healthcheck: + test: ["CMD", "curl", "-f", "http://localhost:3111/agentmemory/livez"] + interval: 5s + timeout: 3s + retries: 10 + start_period: 5s + restart: unless-stopped + logging: + driver: json-file + options: + max-size: "10m" + max-file: "3" + + agentmemory: + build: + context: . + dockerfile: Dockerfile + args: + III_VERSION: ${AGENTMEMORY_III_VERSION:-0.11.2} + depends_on: + iii-engine: + condition: service_healthy + ports: + - "127.0.0.1:3114:3114" + - "127.0.0.1:3113:3113" + volumes: + - iii-data:/data + environment: + AGENTMEMORY_TOOLS: all + AGENTMEMORY_SECRET_FILE: /data/.hmac + III_ENGINE_URL: ws://iii-engine:49134 + III_REST_PORT: "3111" + AGENTMEMORY_MCP_PORT: "3114" + healthcheck: + test: ["CMD", "curl", "-f", "http://localhost:3111/agentmemory/livez"] + interval: 10s + timeout: 5s + retries: 5 + start_period: 20s restart: unless-stopped logging: driver: json-file diff --git a/entrypoint.sh b/entrypoint.sh new file mode 100755 index 000000000..a9fa58d00 --- /dev/null +++ b/entrypoint.sh @@ -0,0 +1,78 @@ +#!/bin/sh +set -eu + +DATA_DIR="${AGENTMEMORY_DATA_DIR:-/data}" +HMAC_FILE="${AGENTMEMORY_HMAC_FILE:-/data/.hmac}" + +mkdir -p "$DATA_DIR" +chown -R node:node "$DATA_DIR" + +if [ ! -s "$HMAC_FILE" ]; then + SECRET="$(openssl rand -hex 32)" + umask 077 + printf '%s\n' "$SECRET" > "$HMAC_FILE" + chmod 600 "$HMAC_FILE" + chown node:node "$HMAC_FILE" + echo "Generated new AGENTMEMORY_SECRET" +fi + +if [ -z "${AGENTMEMORY_SECRET_FILE:-}" ]; then + export AGENTMEMORY_SECRET_FILE="$HMAC_FILE" +fi + +cat > /app/iii-config.yaml <<'EOF' +workers: + - name: iii-http + config: + port: 3111 + host: 0.0.0.0 + default_timeout: 180000 + cors: + allowed_origins: + - "http://localhost:3111" + - "http://localhost:3113" + - "http://localhost:3114" + - "http://127.0.0.1:3111" + - "http://127.0.0.1:3113" + - "http://127.0.0.1:3114" + allowed_methods: [GET, POST, PUT, DELETE, OPTIONS] + - name: iii-state + config: + adapter: + name: kv + config: + store_method: file_based + file_path: /data/state_store.db + - name: iii-queue + config: + adapter: + name: builtin + - name: iii-pubsub + config: + adapter: + name: local + - name: iii-cron + config: + adapter: + name: kv + - name: iii-stream + config: + port: 3112 + host: 0.0.0.0 + adapter: + name: kv + config: + store_method: file_based + file_path: /data/stream_store + - name: iii-observability + config: + enabled: true + service_name: agentmemory + exporter: memory + sampling_ratio: 0.1 + metrics_enabled: true + logs_enabled: true + logs_console_output: false +EOF + +exec node /app/dist/index.mjs diff --git a/iii-config.docker.yaml b/iii-config.docker.yaml index 682236ec3..2889503be 100644 --- a/iii-config.docker.yaml +++ b/iii-config.docker.yaml @@ -5,7 +5,7 @@ workers: host: 0.0.0.0 default_timeout: 180000 cors: - allowed_origins: ["http://localhost:3111", "http://localhost:3113", "http://127.0.0.1:3111", "http://127.0.0.1:3113"] + allowed_origins: ["http://localhost:3111", "http://localhost:3113", "http://localhost:3114", "http://127.0.0.1:3111", "http://127.0.0.1:3113", "http://127.0.0.1:3114"] allowed_methods: [GET, POST, PUT, DELETE, OPTIONS] - name: iii-state config: @@ -45,9 +45,3 @@ workers: metrics_enabled: true logs_enabled: true logs_console_output: false - - name: iii-exec - config: - watch: - - src/**/*.ts - exec: - - node dist/index.mjs diff --git a/plugin/scripts/notification.mjs b/plugin/scripts/notification.mjs index 3967158c9..95786fb40 100755 --- a/plugin/scripts/notification.mjs +++ b/plugin/scripts/notification.mjs @@ -1,7 +1,6 @@ #!/usr/bin/env node import { execSync } from "node:child_process"; import { basename } from "node:path"; - //#region src/hooks/_project.ts function resolveProject(cwd) { const explicit = process.env["AGENTMEMORY_PROJECT_NAME"]; @@ -21,7 +20,6 @@ function resolveProject(cwd) { } catch {} return basename(dir); } - //#endregion //#region src/hooks/notification.ts function isSdkChildContext(payload) { @@ -70,7 +68,7 @@ async function main() { setTimeout(() => process.exit(0), 500).unref(); } main(); - //#endregion -export { }; +export {}; + //# sourceMappingURL=notification.mjs.map \ No newline at end of file diff --git a/plugin/scripts/post-commit.mjs b/plugin/scripts/post-commit.mjs index 8552cd614..5318184f8 100755 --- a/plugin/scripts/post-commit.mjs +++ b/plugin/scripts/post-commit.mjs @@ -1,7 +1,6 @@ #!/usr/bin/env node import { execFile } from "node:child_process"; import { promisify } from "node:util"; - //#region src/hooks/post-commit.ts const exec = promisify(execFile); function isSdkChildContext(payload) { @@ -96,7 +95,7 @@ async function main() { } catch {} } main(); - //#endregion -export { }; +export {}; + //# sourceMappingURL=post-commit.mjs.map \ No newline at end of file diff --git a/plugin/scripts/post-tool-failure.mjs b/plugin/scripts/post-tool-failure.mjs index 6fdad8d9d..4562a7474 100755 --- a/plugin/scripts/post-tool-failure.mjs +++ b/plugin/scripts/post-tool-failure.mjs @@ -1,7 +1,6 @@ #!/usr/bin/env node import { execSync } from "node:child_process"; import { basename } from "node:path"; - //#region src/hooks/_project.ts function resolveProject(cwd) { const explicit = process.env["AGENTMEMORY_PROJECT_NAME"]; @@ -21,7 +20,6 @@ function resolveProject(cwd) { } catch {} return basename(dir); } - //#endregion //#region src/hooks/post-tool-failure.ts function isSdkChildContext(payload) { @@ -71,7 +69,7 @@ async function main() { setTimeout(() => process.exit(0), 500).unref(); } main(); - //#endregion -export { }; +export {}; + //# sourceMappingURL=post-tool-failure.mjs.map \ No newline at end of file diff --git a/plugin/scripts/post-tool-use.mjs b/plugin/scripts/post-tool-use.mjs index b4aef9c94..1103a54f3 100755 --- a/plugin/scripts/post-tool-use.mjs +++ b/plugin/scripts/post-tool-use.mjs @@ -1,7 +1,6 @@ #!/usr/bin/env node import { execSync } from "node:child_process"; import { basename } from "node:path"; - //#region src/hooks/_project.ts function resolveProject(cwd) { const explicit = process.env["AGENTMEMORY_PROJECT_NAME"]; @@ -21,7 +20,6 @@ function resolveProject(cwd) { } catch {} return basename(dir); } - //#endregion //#region src/hooks/post-tool-use.ts function isSdkChildContext(payload) { @@ -116,7 +114,7 @@ function truncate(value, max) { return value; } main(); - //#endregion -export { }; +export {}; + //# sourceMappingURL=post-tool-use.mjs.map \ No newline at end of file diff --git a/plugin/scripts/pre-compact.mjs b/plugin/scripts/pre-compact.mjs index 0afdcb0b4..bf7ce2b76 100755 --- a/plugin/scripts/pre-compact.mjs +++ b/plugin/scripts/pre-compact.mjs @@ -1,7 +1,6 @@ #!/usr/bin/env node import { execSync } from "node:child_process"; import { basename } from "node:path"; - //#region src/hooks/_project.ts function resolveProject(cwd) { const explicit = process.env["AGENTMEMORY_PROJECT_NAME"]; @@ -21,7 +20,6 @@ function resolveProject(cwd) { } catch {} return basename(dir); } - //#endregion //#region src/hooks/pre-compact.ts function isSdkChildContext(payload) { @@ -74,7 +72,7 @@ async function main() { } catch {} } main(); - //#endregion -export { }; +export {}; + //# sourceMappingURL=pre-compact.mjs.map \ No newline at end of file diff --git a/plugin/scripts/pre-tool-use.mjs b/plugin/scripts/pre-tool-use.mjs index d70c166ed..a0d6e9583 100755 --- a/plugin/scripts/pre-tool-use.mjs +++ b/plugin/scripts/pre-tool-use.mjs @@ -78,7 +78,7 @@ async function main() { } catch {} } main(); - //#endregion -export { }; +export {}; + //# sourceMappingURL=pre-tool-use.mjs.map \ No newline at end of file diff --git a/plugin/scripts/prompt-submit.mjs b/plugin/scripts/prompt-submit.mjs index 1a4147e6a..112bcbcc3 100755 --- a/plugin/scripts/prompt-submit.mjs +++ b/plugin/scripts/prompt-submit.mjs @@ -1,7 +1,6 @@ #!/usr/bin/env node import { execSync } from "node:child_process"; import { basename } from "node:path"; - //#region src/hooks/_project.ts function resolveProject(cwd) { const explicit = process.env["AGENTMEMORY_PROJECT_NAME"]; @@ -21,7 +20,6 @@ function resolveProject(cwd) { } catch {} return basename(dir); } - //#endregion //#region src/hooks/prompt-submit.ts function isSdkChildContext(payload) { @@ -63,7 +61,7 @@ async function main() { setTimeout(() => process.exit(0), 500).unref(); } main(); - //#endregion -export { }; +export {}; + //# sourceMappingURL=prompt-submit.mjs.map \ No newline at end of file diff --git a/plugin/scripts/session-end.mjs b/plugin/scripts/session-end.mjs index 019149c30..0ba564be5 100755 --- a/plugin/scripts/session-end.mjs +++ b/plugin/scripts/session-end.mjs @@ -54,7 +54,7 @@ async function main() { setTimeout(() => process.exit(0), 1500).unref(); } main(); - //#endregion -export { }; +export {}; + //# sourceMappingURL=session-end.mjs.map \ No newline at end of file diff --git a/plugin/scripts/session-start.mjs b/plugin/scripts/session-start.mjs index 51b70eb4c..bad7ffc0b 100755 --- a/plugin/scripts/session-start.mjs +++ b/plugin/scripts/session-start.mjs @@ -1,7 +1,6 @@ #!/usr/bin/env node import { execSync } from "node:child_process"; import { basename } from "node:path"; - //#region src/hooks/_project.ts function resolveProject(cwd) { const explicit = process.env["AGENTMEMORY_PROJECT_NAME"]; @@ -21,7 +20,6 @@ function resolveProject(cwd) { } catch {} return basename(dir); } - //#endregion //#region src/hooks/session-start.ts function isSdkChildContext(payload) { @@ -81,7 +79,7 @@ async function main() { } catch {} } main(); - //#endregion -export { }; +export {}; + //# sourceMappingURL=session-start.mjs.map \ No newline at end of file diff --git a/plugin/scripts/stop.mjs b/plugin/scripts/stop.mjs index 03d30c64f..63051c830 100755 --- a/plugin/scripts/stop.mjs +++ b/plugin/scripts/stop.mjs @@ -38,7 +38,7 @@ async function main() { setTimeout(() => process.exit(0), 1500).unref(); } main(); - //#endregion -export { }; +export {}; + //# sourceMappingURL=stop.mjs.map \ No newline at end of file diff --git a/plugin/scripts/subagent-start.mjs b/plugin/scripts/subagent-start.mjs index 2359a1c62..685b78349 100755 --- a/plugin/scripts/subagent-start.mjs +++ b/plugin/scripts/subagent-start.mjs @@ -1,7 +1,6 @@ #!/usr/bin/env node import { execSync } from "node:child_process"; import { basename } from "node:path"; - //#region src/hooks/_project.ts function resolveProject(cwd) { const explicit = process.env["AGENTMEMORY_PROJECT_NAME"]; @@ -21,7 +20,6 @@ function resolveProject(cwd) { } catch {} return basename(dir); } - //#endregion //#region src/hooks/subagent-start.ts function isSdkChildContext(payload) { @@ -69,7 +67,7 @@ async function main() { setTimeout(() => process.exit(0), 500).unref(); } main(); - //#endregion -export { }; +export {}; + //# sourceMappingURL=subagent-start.mjs.map \ No newline at end of file diff --git a/plugin/scripts/subagent-stop.mjs b/plugin/scripts/subagent-stop.mjs index 2ba1b002c..715d8207c 100755 --- a/plugin/scripts/subagent-stop.mjs +++ b/plugin/scripts/subagent-stop.mjs @@ -1,7 +1,6 @@ #!/usr/bin/env node import { execSync } from "node:child_process"; import { basename } from "node:path"; - //#region src/hooks/_project.ts function resolveProject(cwd) { const explicit = process.env["AGENTMEMORY_PROJECT_NAME"]; @@ -21,7 +20,6 @@ function resolveProject(cwd) { } catch {} return basename(dir); } - //#endregion //#region src/hooks/subagent-stop.ts function isSdkChildContext(payload) { @@ -70,7 +68,7 @@ async function main() { setTimeout(() => process.exit(0), 500).unref(); } main(); - //#endregion -export { }; +export {}; + //# sourceMappingURL=subagent-stop.mjs.map \ No newline at end of file diff --git a/plugin/scripts/task-completed.mjs b/plugin/scripts/task-completed.mjs index 478f0688e..9ca9520e2 100755 --- a/plugin/scripts/task-completed.mjs +++ b/plugin/scripts/task-completed.mjs @@ -1,7 +1,6 @@ #!/usr/bin/env node import { execSync } from "node:child_process"; import { basename } from "node:path"; - //#region src/hooks/_project.ts function resolveProject(cwd) { const explicit = process.env["AGENTMEMORY_PROJECT_NAME"]; @@ -21,7 +20,6 @@ function resolveProject(cwd) { } catch {} return basename(dir); } - //#endregion //#region src/hooks/task-completed.ts function isSdkChildContext(payload) { @@ -69,7 +67,7 @@ async function main() { setTimeout(() => process.exit(0), 500).unref(); } main(); - //#endregion -export { }; +export {}; + //# sourceMappingURL=task-completed.mjs.map \ No newline at end of file diff --git a/specs/001-mcp-stream-dockerize/spec.md b/specs/001-mcp-stream-dockerize/spec.md new file mode 100644 index 000000000..857a18e7a --- /dev/null +++ b/specs/001-mcp-stream-dockerize/spec.md @@ -0,0 +1,125 @@ +# Feature Specification: MCP Stream HTTP + Dockerization + +**Feature Branch**: `001-mcp-stream-dockerize` + +**Created**: 2026-06-14 + +**Status**: Draft + +**Input**: User description: "我有两个需求：一，给现在这个project的MCP加上Stream HTTP或SSE的支持，因为是在本地的网络运行，如果Stream HTTP必须要HTTPS的话，就走SSE，如果Stream HTTP可以不走HTTPS，用HTTP就可以的吧，就支持Stream HTTP。二，要求Dockerize这个。可以一个docker compose运行，并且要求完整的Docker image，不能运行把本地的项目路径mount进行运行。如果可以编译成二进制文件在Docker里运行最好，不行的话，在Docker里用python3运行也可以。" + +## Clarifications + +### Session 2026-06-14 + +- Q: Should the Stream HTTP endpoint require authentication? → A: Same Bearer token (`AGENTMEMORY_SECRET`) as the existing REST API. +- Q: Pre-built registry image or local build? → A: Local build via `docker compose build` — no registry publishing required for initial release. +- Q: Keep stdio transport in Docker image? → A: Yes — Docker image supports both stdio (`docker run` as MCP subprocess) and HTTP URL connections. +- Q: Default tool count in Docker? → A: All 53 tools (`AGENTMEMORY_TOOLS=all`) as the Docker compose default for the full experience. +- Q: How should AGENTMEMORY_SECRET be set in Docker? → A: Auto-generate on first boot (random secret), persist to `/data/.hmac` volume, display hint in startup logs. +- Q: [SDK research] Can Stream HTTP work over plain HTTP without SSE? → A: Yes — hand-rolled JSON-RPC server over Node.js `http` module, avoiding the `@modelcontextprotocol/sdk` body-parsing bug with chunked transfer encoding. SSE fallback is unnecessary. + +## User Scenarios & Testing *(mandatory)* + +### User Story 1 - MCP Stream HTTP Transport (Priority: P1) + +An MCP client (such as Claude Code, Cursor, or any MCP-compatible host) connects to agentmemory over HTTP to list and call memory tools. Instead of being limited to stdio (subprocess-based) connections, the client opens an HTTP connection to the agentmemory server and receives responses via a streaming transport — the server processes tool calls and streams results back incrementally as they become available, without waiting for the complete response. Since the deployment runs on a local network, the transport operates over plain HTTP (not HTTPS). + +**Why this priority**: This is the foundational change — without it, MCP clients are restricted to stdio connections, which prevents remote/containerized deployments and limits the integration options available to users. All existing MCP tool functionality (memory CRUD, search, sessions, etc.) must be accessible through this new transport. + +**Independent Test**: Start the agentmemory server with Stream HTTP enabled, send an MCP `initialize` request via `curl` to the stream endpoint, receive a valid JSON-RPC response with server capabilities, then call `tools/list` and `tools/call` over the same HTTP connection. Verify all 53 tools are listed and callable. + +**Acceptance Scenarios**: + +1. **Given** agentmemory server is running with Stream HTTP transport enabled, **When** an MCP client sends a JSON-RPC `initialize` request to the stream endpoint, **Then** the server responds with protocol version and capabilities including `tools: {}` support. +2. **Given** an initialized MCP session over Stream HTTP, **When** the client calls `tools/list`, **Then** the server returns the full list of available tools (53 tools with all features enabled). +3. **Given** an initialized MCP session over Stream HTTP, **When** the client calls `tools/call` with tool name `memory_search` and valid arguments, **Then** the server executes the search and returns the results as a JSON-RPC response. +4. **Given** both Stream HTTP and stdio transports are available, **When** a client connects via either transport, **Then** both transports expose the identical set of tools with identical behavior. + +--- + +### User Story 2 - Docker Compose Deployment (Priority: P2) + +A user wants to run agentmemory entirely in containers without cloning the source repository. They run a single `docker compose up` command, and the full stack starts — the agentmemory application and its iii-engine dependency — all from pre-built Docker images. No source code directories are bind-mounted into the containers; everything is baked into the images. + +**Why this priority**: Dockerization makes agentmemory deployable anywhere Docker runs, without Node.js or npm prerequisites. This unlocks production deployments, CI/CD integration, and self-hosted usage. + +**Independent Test**: On a machine with only Docker and docker-compose installed (no Node.js, no cloned repo), run `docker compose up`, wait for the services to report healthy, then `curl http://localhost:3111/agentmemory/livez` and receive `{"status": "ok"}`. Then use an MCP client to connect to the Stream HTTP endpoint and perform a memory save and recall. + +**Acceptance Scenarios**: + +1. **Given** Docker and docker-compose are installed, **When** the user runs `docker compose up`, **Then** both the iii-engine and agentmemory containers start and reach a healthy state. +2. **Given** the stack is running via docker compose, **When** the user sends a request to the MCP Stream HTTP endpoint, **Then** the request is processed successfully with streaming responses. +3. **Given** the agentmemory container, **When** inspected, **Then** no source code directories are bind-mounted; the application and all dependencies are self-contained in the image. +4. **Given** the docker compose stack is stopped and restarted, **When** the user runs `docker compose down` then `docker compose up`, **Then** previously stored memory data persists (via a named Docker volume). + +--- + +### User Story 3 - Single-Command Setup for New Users (Priority: P3) + +A new user with no prior agentmemory installation wants to get started. They download the `docker-compose.yml` file, run `docker compose up`, and within seconds the service is ready. They configure their MCP client with a single HTTP URL (no subprocess paths, no `npx` commands) and immediately have working memory in their AI coding sessions. + +**Why this priority**: Streamlined onboarding removes friction for adoption. This is the "it just works" story that turns a complex multi-step setup into two commands. + +**Independent Test**: On a fresh machine, create a new directory, download `docker-compose.yml`, run `docker compose up`, then configure any MCP client with `http://localhost:/mcp` as the connection URL and verify memory tools are available and functional. + +**Acceptance Scenarios**: + +1. **Given** a fresh environment with only Docker, **When** the user obtains the `docker-compose.yml` and runs `docker compose up`, **Then** within 30 seconds the service is ready to accept MCP connections. +2. **Given** the server is running in Docker, **When** the user follows setup instructions to connect their MCP client via HTTP URL, **Then** the client successfully connects and lists all memory tools. +3. **Given** the server is running in Docker, **When** the user checks the Docker image size, **Then** the agentmemory image is reasonably compact and optimized (no unnecessary build tools, no source code, no `node_modules` bloat). + +--- + +### Edge Cases + +- What happens when the iii-engine container starts after the agentmemory container? The agentmemory container must retry the engine connection with backoff until the engine is ready. +- How does the system handle concurrent MCP clients over Stream HTTP? Multiple clients must be able to maintain independent sessions simultaneously. +- What happens when a Stream HTTP connection is dropped mid-stream? The server must clean up session resources and not leak connections. +- How does the server behave when a client sends an invalid JSON-RPC message? Return a proper JSON-RPC error response without crashing. +- What happens to stored memory data when the Docker containers are rebuilt or updated? Data persists in the Docker volume and survives image updates. +- How does the container handle graceful shutdown when `docker compose down` is issued? Connections must close cleanly and data must be flushed. + +## Requirements *(mandatory)* + +### Functional Requirements + +- **FR-001**: The MCP server MUST support the Streamable HTTP transport as defined by the MCP specification, operating over plain HTTP — using a hand-rolled JSON-RPC server for direct JSON responses. +- **FR-002**: The new HTTP-based MCP transport MUST expose all existing tools (53 tools with `AGENTMEMORY_TOOLS=all`) with identical inputs and outputs as the existing stdio transport. +- **FR-003**: The existing stdio transport (JSON-RPC 2.0 over stdin/stdout) MUST continue to function alongside the new HTTP transport — both transports must coexist, including when running from within the Docker container. +- **FR-004**: The system MUST provide a complete Docker image for the agentmemory application that contains all runtime dependencies and the built application — no source code or `node_modules` directory mount from the host. +- **FR-005**: The system MUST provide a `docker-compose.yml` that defines and orchestrates both the agentmemory and iii-engine services, including correct startup ordering and health checks. +- **FR-006**: The Docker deployment MUST persist memory data in a named Docker volume, surviving container restarts and image updates. +- **FR-007**: The agentmemory container MUST gracefully handle the iii-engine container starting after it, with connection retries. +- **FR-008**: The MCP Stream HTTP endpoint MUST support concurrent client connections, each with independent session state. +- **FR-009**: Graceful shutdown MUST be handled — connections closed and data flushed on termination signals. +- **FR-010**: The Stream HTTP endpoint MUST require the same Bearer token authentication (`AGENTMEMORY_SECRET`) as the existing REST API, ensuring consistent access control across all transports. +- **FR-011**: The docker-compose.yml MUST default to exposing all 53 MCP tools (`AGENTMEMORY_TOOLS=all`) for a complete out-of-box experience. +- **FR-012**: On first container startup, if `AGENTMEMORY_SECRET` is not set, the container MUST auto-generate a random secret, persist it to a file on the data volume, and log instructions for retrieving it. + +### Key Entities + +- **MCP Session**: Represents an active client connection over the Stream HTTP transport. Key attributes: session identifier, initialization state, client capabilities, negotiated protocol version. +- **Docker Service (agentmemory)**: The containerized agentmemory application. Depends on the iii-engine service. Exposes MCP and API ports. +- **Docker Service (iii-engine)**: The containerized iii-engine backend. Provides state storage, pub/sub, and HTTP worker infrastructure to agentmemory. + +## Success Criteria *(mandatory)* + +### Measurable Outcomes + +- **SC-001**: An MCP client can connect to agentmemory over HTTP, complete initialization, list tools, and call any tool — all without using stdio or spawning a subprocess. +- **SC-002**: All 53 MCP tools are accessible and functional through the new HTTP-based transport. +- **SC-003**: A user with only Docker installed can go from zero to a running agentmemory instance with a single `docker compose up` command. +- **SC-004**: The agentmemory Docker image contains no source code and requires no host directory mounts to function. +- **SC-005**: Memory data persists across `docker compose down && docker compose up` cycles. +- **SC-006**: The system handles at least 5 concurrent MCP client connections over Stream HTTP without errors or throughput degradation. +- **SC-007**: The existing stdio-based MCP transport continues to pass all existing tests after the HTTP transport is added. + +## Assumptions + +- The MCP Streamable HTTP transport is implemented as a hand-rolled JSON-RPC server using Node.js `http` module (not using `@modelcontextprotocol/sdk`) because the SDK's `NodeStreamableHTTPServerTransport` has a body-parsing bug with chunked transfer encoding. +- The `iii-engine` Docker image (`iiidev/iii:0.11.2`) remains the dependency — this feature does not modify or rebuild the iii-engine. +- The Stream HTTP endpoint (`POST /mcp`) is served on port 3114, separate from the existing REST API on port 3111. +- The existing deploy templates in `deploy/` (Fly, Railway, Render, Coolify) are out of scope for this feature — only a new Docker Compose setup targeting local/Docker-host deployments is required. +- Node.js is the runtime for the Docker image. Binary compilation via tools like Bun or `pkg` may be explored but is not required. +- The Docker image is built locally via `docker compose build` — no container registry publishing is required for the initial release. The `docker-compose.yml` includes the build context and Dockerfile reference. diff --git a/src/cli/connect/json-mcp-adapter.ts b/src/cli/connect/json-mcp-adapter.ts index 35998e0dc..b58f951b1 100644 --- a/src/cli/connect/json-mcp-adapter.ts +++ b/src/cli/connect/json-mcp-adapter.ts @@ -12,6 +12,15 @@ import { writeJsonAtomic, } from "./util.js"; +// MCP transport modes: +// - "command" (default): stdio transport — npx @agentmemory/mcp spawns a +// child process; clients speak JSON-RPC over stdin/stdout. This is what +// every adapter here writes to .mcp.json today. +// - "url": Streamable HTTP transport — clients connect to the MCP server's +// HTTP endpoint directly (http://localhost:3114/mcp) with a Bearer token. +// Clients that support the MCP Streamable HTTP spec can use this instead +// of the npx command. See AGENTMEMORY_MCP_HTTP_BLOCK in util.ts. + export type JsonMcpAdapterConfig = { name: string; displayName: string; diff --git a/src/cli/connect/types.ts b/src/cli/connect/types.ts index 169dde138..21f25c381 100644 --- a/src/cli/connect/types.ts +++ b/src/cli/connect/types.ts @@ -27,7 +27,7 @@ export type ConnectAdapter = { * server only. Declared on the adapter so the picker never needs a * separate hardcoded list (#872). Defaults to "mcp" when omitted. */ - category?: "native" | "mcp"; + category?: "native" | "mcp" | "stream-http"; detect(): boolean; install(opts: ConnectOptions): Promise; }; diff --git a/src/cli/connect/util.ts b/src/cli/connect/util.ts index 580cd4ee7..0b3ae3edf 100644 --- a/src/cli/connect/util.ts +++ b/src/cli/connect/util.ts @@ -43,6 +43,19 @@ const COPILOT_MCP_COMMAND = args: ["-y", "@agentmemory/mcp"], }; +// Stream HTTP MCP entry (alternative to stdio command): +// clients that speak the Streamable HTTP transport (MCP spec 2025-03-26+) +// can connect directly to the agentmemory MCP port at :3114 instead of +// spawning an npx process. Use this block (or merge it with the stdio +// "agentmemory" entry as an additional server entry) when the MCP server +// is reachable as an HTTP endpoint — no npx bootstrap or stdio needed. +export const AGENTMEMORY_MCP_HTTP_BLOCK = { + url: "http://localhost:3114/mcp", + headers: { + Authorization: "Bearer ${AGENTMEMORY_SECRET:-}", + }, +}; + export const AGENTMEMORY_COPILOT_MCP_BLOCK = { type: "local" as const, ...COPILOT_MCP_COMMAND, diff --git a/src/config.ts b/src/config.ts index f68da2e31..11183ed77 100644 --- a/src/config.ts +++ b/src/config.ts @@ -161,7 +161,7 @@ export function loadConfig(): AgentMemoryConfig { const provider = detectProvider(env); - // Port quartet: REST is the anchor; streams/engine derive from it + // Port family: REST is the anchor — streams, MCP, and engine ports derive from it // unless individually overridden. Default anchor 3111 yields the // canonical 3112 streams / 49134 engine pair, but `III_REST_PORT=3211` // auto-picks 3212 + 49234 so a second instance doesn't collide (#750). @@ -169,6 +169,8 @@ export function loadConfig(): AgentMemoryConfig { const streamsPort = parseInt(env["III_STREAM_PORT"] || env["III_STREAMS_PORT"] || "", 10) || restPort + 1; + const mcpPort = + parseInt(env["AGENTMEMORY_MCP_PORT"] || "", 10) || restPort + 3; const engineUrl = env["III_ENGINE_URL"] || `ws://localhost:${ @@ -179,6 +181,7 @@ export function loadConfig(): AgentMemoryConfig { engineUrl, restPort, streamsPort, + mcpPort, provider, tokenBudget: safeParseInt(env["TOKEN_BUDGET"], 2000), maxObservationsPerSession: safeParseInt(env["MAX_OBS_PER_SESSION"], 500), diff --git a/src/index.ts b/src/index.ts index 4233e8a67..0ff5de15a 100644 --- a/src/index.ts +++ b/src/index.ts @@ -93,6 +93,7 @@ import { registerEventTriggers } from "./triggers/events.js"; import { registerMcpEndpoints } from "./mcp/server.js"; import { getAllTools } from "./mcp/tools-registry.js"; import { startViewerServer } from "./viewer/server.js"; +import { startMcpStreamServer } from "./mcp/stream-http.js"; import { MetricsStore } from "./eval/metrics-store.js"; import { DedupMap } from "./functions/dedup.js"; import { registerHealthMonitor } from "./health/monitor.js"; @@ -520,6 +521,9 @@ async function main() { bootLog( `REST API: 128 endpoints at http://localhost:${config.restPort}/agentmemory/*`, ); + bootLog( + `MCP Stream HTTP: http://localhost:${config.mcpPort}/mcp (JSON-RPC)`, + ); bootLog( `MCP surface (opt-in via \`npx @agentmemory/mcp\`): ${getAllTools().length} tools · 6 resources · 3 prompts`, ); @@ -533,6 +537,16 @@ async function main() { config.restPort, ); + const mcpServer = await startMcpStreamServer( + config.mcpPort, + sdk, + kv, + secret, + ); + bootLog( + `MCP Stream HTTP server started on port ${config.mcpPort}`, + ); + const autoForgetIntervalMs = parseInt(process.env.AUTO_FORGET_INTERVAL_MS || "3600000", 10); const consolidationIntervalMs = parseInt(process.env.CONSOLIDATION_INTERVAL_MS || "7200000", 10); @@ -595,6 +609,7 @@ async function main() { dedupMap.stop(); indexPersistence.stop(); await new Promise((resolve) => viewerServer.close(() => resolve())); + await mcpServer.shutdown(); await indexPersistence.save().catch((err) => { console.warn(`[agentmemory] Failed to save index on shutdown:`, err); }); diff --git a/src/mcp/handler.ts b/src/mcp/handler.ts new file mode 100644 index 000000000..3653ca4f7 --- /dev/null +++ b/src/mcp/handler.ts @@ -0,0 +1,432 @@ +import type { ISdk } from "iii-sdk"; +import { InMemoryKV } from "./in-memory-kv.js"; +import { getAllTools } from "./tools-registry.js"; +import { VERSION } from "../version.js"; +import { generateId } from "../state/schema.js"; +import { + resolveHandle, + invalidateHandle, + type Handle, + type ProxyHandle, +} from "./rest-proxy.js"; + +const IMPLEMENTED_TOOLS = new Set([ + "memory_save", + "memory_recall", + "memory_smart_search", + "memory_sessions", + "memory_export", + "memory_audit", + "memory_governance_delete", +]); + +let modeAnnounced = false; + +function displayAgentmemoryUrl(): string { + const raw = process.env["AGENTMEMORY_URL"]; + if (!raw || (raw.startsWith("${") && raw.endsWith("}"))) { + return "http://localhost:3111"; + } + return raw; +} + +function announceMode(handle: Handle): void { + if (modeAnnounced) return; + modeAnnounced = true; + if (handle.mode === "proxy") { + process.stderr.write( + `[@agentmemory/mcp] proxying to agentmemory server at ${handle.baseUrl}\n`, + ); + } else { + const fullToolCount = getAllTools().length; + process.stderr.write( + `[@agentmemory/mcp] no server reachable at ${displayAgentmemoryUrl()}; running reduced LOCAL FALLBACK with ${IMPLEMENTED_TOOLS.size} of ${fullToolCount} tools. Start 'npx @agentmemory/agentmemory' (and point AGENTMEMORY_URL at it) to unlock all ${fullToolCount} tools.\n`, + ); + } +} + +function normalizeList(value: unknown): string[] { + if (!value) return []; + if (Array.isArray(value)) { + return value + .map((v) => (typeof v === "string" ? v.trim() : "")) + .filter((v) => v.length > 0); + } + if (typeof value === "string") { + return value + .split(",") + .map((s) => s.trim()) + .filter((s) => s.length > 0); + } + return []; +} + +const DEFAULT_LIMIT = 10; +const MAX_LIMIT = 100; +function parseLimit(raw: unknown, fallback = DEFAULT_LIMIT): number { + if (typeof raw !== "number" && typeof raw !== "string") return fallback; + const n = Number(raw); + if (!Number.isFinite(n) || n <= 0) return fallback; + return Math.min(Math.floor(n), MAX_LIMIT); +} + +function textResponse(payload: unknown, pretty = false): { + content: Array<{ type: string; text: string }>; +} { + return { + content: [ + { type: "text", text: JSON.stringify(payload, null, pretty ? 2 : 0) }, + ], + }; +} + +interface Validated { + tool: string; + content?: string; + type?: string; + concepts?: string[]; + files?: string[]; + query?: string; + limit?: number; + format?: string; + tokenBudget?: number; + memoryIds?: string[]; + reason?: string; +} + +function validate(toolName: string, args: Record): Validated { + if (!IMPLEMENTED_TOOLS.has(toolName)) { + throw new Error(`Unknown tool: ${toolName}`); + } + const v: Validated = { tool: toolName }; + switch (toolName) { + case "memory_save": { + const content = args["content"]; + if (typeof content !== "string" || !content.trim()) { + throw new Error("content is required"); + } + v.content = content; + v.type = (args["type"] as string) || "fact"; + v.concepts = normalizeList(args["concepts"]); + v.files = normalizeList(args["files"]); + return v; + } + case "memory_recall": + case "memory_smart_search": { + const query = args["query"]; + if (typeof query !== "string" || !query.trim()) { + throw new Error("query is required"); + } + v.query = query.trim(); + v.limit = parseLimit(args["limit"]); + const fmt = args["format"]; + if (typeof fmt === "string" && fmt.trim()) { + v.format = fmt.trim().toLowerCase(); + } + const budget = args["token_budget"]; + if (typeof budget === "number" && Number.isFinite(budget) && budget > 0) { + v.tokenBudget = Math.floor(budget); + } else if (typeof budget === "string" && budget.trim()) { + const n = Number(budget); + if (Number.isFinite(n) && n > 0) v.tokenBudget = Math.floor(n); + } + return v; + } + case "memory_sessions": { + v.limit = parseLimit(args["limit"], 20); + return v; + } + case "memory_governance_delete": { + const ids = normalizeList(args["memoryIds"]); + if (ids.length === 0) throw new Error("memoryIds is required"); + v.memoryIds = ids; + v.reason = (args["reason"] as string) || "plugin skill request"; + return v; + } + case "memory_export": + return v; + case "memory_audit": { + v.limit = parseLimit(args["limit"], 50); + return v; + } + default: + throw new Error(`Unknown tool: ${toolName}`); + } +} + +async function handleProxy( + v: Validated, + handle: ProxyHandle, +): Promise<{ content: Array<{ type: string; text: string }> }> { + switch (v.tool) { + case "memory_save": { + const result = await handle.call("/agentmemory/remember", { + method: "POST", + body: JSON.stringify({ + content: v.content, + type: v.type, + concepts: v.concepts, + files: v.files, + }), + }); + return textResponse(result); + } + case "memory_recall": { + const body: Record = { + query: v.query, + limit: v.limit, + format: v.format ?? "full", + }; + if (v.tokenBudget != null) body["token_budget"] = v.tokenBudget; + const result = await handle.call("/agentmemory/search", { + method: "POST", + body: JSON.stringify(body), + }); + return textResponse(result, true); + } + case "memory_smart_search": { + const body: Record = { query: v.query, limit: v.limit }; + if (v.format != null) body["format"] = v.format; + if (v.tokenBudget != null) body["token_budget"] = v.tokenBudget; + const result = await handle.call("/agentmemory/smart-search", { + method: "POST", + body: JSON.stringify(body), + }); + return textResponse(result, true); + } + case "memory_sessions": { + const result = await handle.call( + `/agentmemory/sessions?limit=${v.limit}`, + { method: "GET" }, + ); + return textResponse(result, true); + } + case "memory_governance_delete": { + const result = await handle.call("/agentmemory/governance/memories", { + method: "DELETE", + body: JSON.stringify({ memoryIds: v.memoryIds, reason: v.reason }), + }); + return textResponse(result); + } + case "memory_export": { + const result = await handle.call("/agentmemory/export", { method: "GET" }); + return textResponse(result, true); + } + case "memory_audit": { + const result = await handle.call( + `/agentmemory/audit?limit=${v.limit}`, + { method: "GET" }, + ); + return textResponse(result, true); + } + default: + throw new Error(`Unknown tool: ${v.tool}`); + } +} + +async function handleLocal( + v: Validated, + kvInstance: InMemoryKV, +): Promise<{ content: Array<{ type: string; text: string }> }> { + switch (v.tool) { + case "memory_save": { + const id = generateId("mem"); + const isoNow = new Date().toISOString(); + await kvInstance.set("mem:memories", id, { + id, + type: v.type, + title: (v.content || "").slice(0, 80), + content: v.content, + concepts: v.concepts, + files: v.files, + createdAt: isoNow, + updatedAt: isoNow, + strength: 7, + version: 1, + isLatest: true, + sessionIds: [], + }); + kvInstance.persist(); + return textResponse({ saved: id }); + } + + case "memory_recall": + case "memory_smart_search": { + const query = (v.query || "").toLowerCase(); + const limit = v.limit ?? DEFAULT_LIMIT; + const all = + await kvInstance.list>("mem:memories"); + const results = all + .filter((m) => { + const text = [ + typeof m["title"] === "string" ? m["title"] : "", + typeof m["content"] === "string" ? m["content"] : "", + Array.isArray(m["files"]) ? m["files"].join(" ") : "", + Array.isArray(m["concepts"]) ? m["concepts"].join(" ") : "", + Array.isArray(m["sessionIds"]) ? m["sessionIds"].join(" ") : "", + typeof m["id"] === "string" ? m["id"] : "", + ] + .join(" ") + .toLowerCase(); + return query.split(/\s+/).every((word) => text.includes(word)); + }) + .slice(0, limit); + return textResponse({ mode: "compact", results }, true); + } + + case "memory_sessions": { + const sessions = + await kvInstance.list>("mem:sessions"); + const limit = v.limit ?? 20; + return textResponse({ sessions: sessions.slice(0, limit) }, true); + } + + case "memory_governance_delete": { + let deleted = 0; + for (const id of v.memoryIds || []) { + const existing = await kvInstance.get("mem:memories", id); + if (existing) { + await kvInstance.delete("mem:memories", id); + deleted++; + } + } + kvInstance.persist(); + return textResponse({ + deleted, + requested: (v.memoryIds || []).length, + reason: v.reason, + }); + } + + case "memory_export": { + const memories = await kvInstance.list("mem:memories"); + const sessions = await kvInstance.list("mem:sessions"); + return textResponse({ version: VERSION, memories, sessions }, true); + } + + case "memory_audit": { + const entries = await kvInstance.list("mem:audit"); + const limit = v.limit ?? 50; + return textResponse( + { + entries: (entries as Array>).slice(0, limit), + }, + true, + ); + } + + default: + throw new Error(`Unknown tool: ${v.tool}`); + } +} + +async function handleProxyGeneric( + toolName: string, + args: Record, + handle: ProxyHandle, +): Promise<{ content: Array<{ type: string; text: string }> }> { + const result = (await handle.call("/agentmemory/mcp/call", { + method: "POST", + body: JSON.stringify({ name: toolName, arguments: args }), + })) as { content?: Array<{ type: string; text: string }> } | null; + if (result && Array.isArray(result.content)) { + return { content: result.content }; + } + return textResponse(result, true); +} + +export async function handleToolCall( + toolName: string, + args: Record, + _sdk: ISdk | null, + kvInstance: InMemoryKV, + _config?: Record, +): Promise<{ content: Array<{ type: string; text: string }> }> { + const handle = await resolveHandle(); + announceMode(handle); + + if (!IMPLEMENTED_TOOLS.has(toolName)) { + if (handle.mode === "proxy") { + try { + return await handleProxyGeneric(toolName, args, handle); + } catch (err) { + process.stderr.write( + `[@agentmemory/mcp] proxy call failed for ${toolName}: ${err instanceof Error ? err.message : String(err)}\n`, + ); + invalidateHandle(); + throw err; + } + } + throw new Error( + `Unknown tool: ${toolName} (local fallback supports only ${[...IMPLEMENTED_TOOLS].join(", ")}; start an agentmemory server and set AGENTMEMORY_URL to use the full tool set)`, + ); + } + + const validated = validate(toolName, args); + if (handle.mode === "proxy") { + try { + return await handleProxy(validated, handle); + } catch (err) { + process.stderr.write( + `[@agentmemory/mcp] proxy call failed for ${toolName}: ${err instanceof Error ? err.message : String(err)}; invalidating handle and falling back to local KV\n`, + ); + invalidateHandle(); + } + } + return handleLocal(validated, kvInstance); +} + +export async function handleToolsList( + _sdk: ISdk | null, + _kv: InMemoryKV, + _config?: Record, +): Promise<{ tools: unknown[] }> { + const debug = process.env["AGENTMEMORY_DEBUG"] === "1" || process.env["AGENTMEMORY_DEBUG"] === "true"; + const handle = await resolveHandle(); + announceMode(handle); + if (debug) { + process.stderr.write( + `[@agentmemory/mcp] tools/list: handle.mode=${handle.mode}${handle.mode === "proxy" ? ` baseUrl=${handle.baseUrl}` : ""}\n`, + ); + } + if (handle.mode === "proxy") { + try { + const remote = (await handle.call("/agentmemory/mcp/tools", { + method: "GET", + })) as { tools?: unknown } | null; + if (debug) { + const shape = remote === null + ? "null" + : typeof remote !== "object" + ? typeof remote + : `keys=${Object.keys(remote as object).join(",")} toolsType=${Array.isArray((remote as { tools?: unknown }).tools) ? `array(len=${((remote as { tools: unknown[] }).tools).length})` : typeof (remote as { tools?: unknown }).tools}`; + process.stderr.write( + `[@agentmemory/mcp] tools/list: remote response shape: ${shape}\n`, + ); + } + if (remote && Array.isArray(remote.tools)) { + if (debug) { + process.stderr.write( + `[@agentmemory/mcp] tools/list: returning ${remote.tools.length} tools from server\n`, + ); + } + return { tools: remote.tools }; + } + process.stderr.write( + `[@agentmemory/mcp] tools/list: server returned unexpected shape (no .tools array); falling back to local IMPLEMENTED_TOOLS list. Set AGENTMEMORY_DEBUG=1 to inspect response.\n`, + ); + } catch (err) { + process.stderr.write( + `[@agentmemory/mcp] tools/list proxy failed: ${err instanceof Error ? err.message : String(err)}; falling back to local list\n`, + ); + invalidateHandle(); + } + } + const fallback = getAllTools().filter((t) => IMPLEMENTED_TOOLS.has(t.name)); + if (debug) { + process.stderr.write( + `[@agentmemory/mcp] tools/list: returning ${fallback.length} local fallback tools (${fallback.map((t) => t.name).join(",")})\n`, + ); + } + return { tools: fallback }; +} diff --git a/src/mcp/standalone.ts b/src/mcp/standalone.ts index dd66ecb1d..16b3895e8 100644 --- a/src/mcp/standalone.ts +++ b/src/mcp/standalone.ts @@ -2,26 +2,9 @@ import { InMemoryKV } from "./in-memory-kv.js"; import { createStdioTransport } from "./transport.js"; -import { getAllTools } from "./tools-registry.js"; import { getStandalonePersistPath } from "../config.js"; import { VERSION } from "../version.js"; -import { generateId } from "../state/schema.js"; -import { - resolveHandle, - invalidateHandle, - type Handle, - type ProxyHandle, -} from "./rest-proxy.js"; - -const IMPLEMENTED_TOOLS = new Set([ - "memory_save", - "memory_recall", - "memory_smart_search", - "memory_sessions", - "memory_export", - "memory_audit", - "memory_governance_delete", -]); +import { handleToolCall as handleToolCallImpl, handleToolsList as handleToolsListImpl } from "./handler.js"; const SERVER_INFO = { name: "agentmemory", @@ -30,419 +13,17 @@ const SERVER_INFO = { }; const kv = new InMemoryKV(getStandalonePersistPath()); -let modeAnnounced = false; - -function displayAgentmemoryUrl(): string { - // Match the literal-placeholder guard in rest-proxy.ts so log lines - // don't show `${AGENTMEMORY_URL}` when an MCP host passed the - // placeholder through unexpanded. - const raw = process.env["AGENTMEMORY_URL"]; - if (!raw || (raw.startsWith("${") && raw.endsWith("}"))) { - return "http://localhost:3111"; - } - return raw; -} - -function announceMode(handle: Handle): void { - if (modeAnnounced) return; - modeAnnounced = true; - if (handle.mode === "proxy") { - process.stderr.write( - `[@agentmemory/mcp] proxying to agentmemory server at ${handle.baseUrl}\n`, - ); - } else { - const fullToolCount = getAllTools().length; - process.stderr.write( - `[@agentmemory/mcp] no server reachable at ${displayAgentmemoryUrl()}; running reduced LOCAL FALLBACK with ${IMPLEMENTED_TOOLS.size} of ${fullToolCount} tools. Start 'npx @agentmemory/agentmemory' (and point AGENTMEMORY_URL at it) to unlock all ${fullToolCount} tools.\n`, - ); - } -} - -function normalizeList(value: unknown): string[] { - if (!value) return []; - if (Array.isArray(value)) { - return value - .map((v) => (typeof v === "string" ? v.trim() : "")) - .filter((v) => v.length > 0); - } - if (typeof value === "string") { - return value - .split(",") - .map((s) => s.trim()) - .filter((s) => s.length > 0); - } - return []; -} - -const DEFAULT_LIMIT = 10; -const MAX_LIMIT = 100; -function parseLimit(raw: unknown, fallback = DEFAULT_LIMIT): number { - if (typeof raw !== "number" && typeof raw !== "string") return fallback; - const n = Number(raw); - if (!Number.isFinite(n) || n <= 0) return fallback; - return Math.min(Math.floor(n), MAX_LIMIT); -} - -function textResponse(payload: unknown, pretty = false): { - content: Array<{ type: string; text: string }>; -} { - return { - content: [ - { type: "text", text: JSON.stringify(payload, null, pretty ? 2 : 0) }, - ], - }; -} - -interface Validated { - tool: string; - content?: string; - type?: string; - concepts?: string[]; - files?: string[]; - query?: string; - limit?: number; - format?: string; - tokenBudget?: number; - memoryIds?: string[]; - reason?: string; -} - -function validate(toolName: string, args: Record): Validated { - if (!IMPLEMENTED_TOOLS.has(toolName)) { - throw new Error(`Unknown tool: ${toolName}`); - } - const v: Validated = { tool: toolName }; - switch (toolName) { - case "memory_save": { - const content = args["content"]; - if (typeof content !== "string" || !content.trim()) { - throw new Error("content is required"); - } - v.content = content; - v.type = (args["type"] as string) || "fact"; - v.concepts = normalizeList(args["concepts"]); - v.files = normalizeList(args["files"]); - return v; - } - case "memory_recall": - case "memory_smart_search": { - const query = args["query"]; - if (typeof query !== "string" || !query.trim()) { - throw new Error("query is required"); - } - v.query = query.trim(); - v.limit = parseLimit(args["limit"]); - const fmt = args["format"]; - if (typeof fmt === "string" && fmt.trim()) { - v.format = fmt.trim().toLowerCase(); - } - const budget = args["token_budget"]; - if (typeof budget === "number" && Number.isFinite(budget) && budget > 0) { - v.tokenBudget = Math.floor(budget); - } else if (typeof budget === "string" && budget.trim()) { - const n = Number(budget); - if (Number.isFinite(n) && n > 0) v.tokenBudget = Math.floor(n); - } - return v; - } - case "memory_sessions": { - v.limit = parseLimit(args["limit"], 20); - return v; - } - case "memory_governance_delete": { - const ids = normalizeList(args["memoryIds"]); - if (ids.length === 0) throw new Error("memoryIds is required"); - v.memoryIds = ids; - v.reason = (args["reason"] as string) || "plugin skill request"; - return v; - } - case "memory_export": - return v; - case "memory_audit": { - v.limit = parseLimit(args["limit"], 50); - return v; - } - default: - throw new Error(`Unknown tool: ${toolName}`); - } -} - -async function handleProxy( - v: Validated, - handle: ProxyHandle, -): Promise<{ content: Array<{ type: string; text: string }> }> { - switch (v.tool) { - case "memory_save": { - const result = await handle.call("/agentmemory/remember", { - method: "POST", - body: JSON.stringify({ - content: v.content, - type: v.type, - concepts: v.concepts, - files: v.files, - }), - }); - return textResponse(result); - } - case "memory_recall": { - const body: Record = { - query: v.query, - limit: v.limit, - format: v.format ?? "full", - }; - if (v.tokenBudget != null) body["token_budget"] = v.tokenBudget; - const result = await handle.call("/agentmemory/search", { - method: "POST", - body: JSON.stringify(body), - }); - return textResponse(result, true); - } - case "memory_smart_search": { - const body: Record = { query: v.query, limit: v.limit }; - if (v.format != null) body["format"] = v.format; - if (v.tokenBudget != null) body["token_budget"] = v.tokenBudget; - const result = await handle.call("/agentmemory/smart-search", { - method: "POST", - body: JSON.stringify(body), - }); - return textResponse(result, true); - } - case "memory_sessions": { - const result = await handle.call( - `/agentmemory/sessions?limit=${v.limit}`, - { method: "GET" }, - ); - return textResponse(result, true); - } - case "memory_governance_delete": { - const result = await handle.call("/agentmemory/governance/memories", { - method: "DELETE", - body: JSON.stringify({ memoryIds: v.memoryIds, reason: v.reason }), - }); - return textResponse(result); - } - case "memory_export": { - const result = await handle.call("/agentmemory/export", { method: "GET" }); - return textResponse(result, true); - } - case "memory_audit": { - const result = await handle.call( - `/agentmemory/audit?limit=${v.limit}`, - { method: "GET" }, - ); - return textResponse(result, true); - } - default: - throw new Error(`Unknown tool: ${v.tool}`); - } -} - -async function handleLocal( - v: Validated, - kvInstance: InMemoryKV, -): Promise<{ content: Array<{ type: string; text: string }> }> { - switch (v.tool) { - case "memory_save": { - const id = generateId("mem"); - const isoNow = new Date().toISOString(); - await kvInstance.set("mem:memories", id, { - id, - type: v.type, - title: (v.content || "").slice(0, 80), - content: v.content, - concepts: v.concepts, - files: v.files, - createdAt: isoNow, - updatedAt: isoNow, - strength: 7, - version: 1, - isLatest: true, - sessionIds: [], - }); - kvInstance.persist(); - return textResponse({ saved: id }); - } - - case "memory_recall": - case "memory_smart_search": { - const query = (v.query || "").toLowerCase(); - const limit = v.limit ?? DEFAULT_LIMIT; - const all = - await kvInstance.list>("mem:memories"); - const results = all - .filter((m) => { - const text = [ - typeof m["title"] === "string" ? m["title"] : "", - typeof m["content"] === "string" ? m["content"] : "", - Array.isArray(m["files"]) ? m["files"].join(" ") : "", - Array.isArray(m["concepts"]) ? m["concepts"].join(" ") : "", - Array.isArray(m["sessionIds"]) ? m["sessionIds"].join(" ") : "", - typeof m["id"] === "string" ? m["id"] : "", - ] - .join(" ") - .toLowerCase(); - return query.split(/\s+/).every((word) => text.includes(word)); - }) - .slice(0, limit); - return textResponse({ mode: "compact", results }, true); - } - - case "memory_sessions": { - const sessions = - await kvInstance.list>("mem:sessions"); - const limit = v.limit ?? 20; - return textResponse({ sessions: sessions.slice(0, limit) }, true); - } - - case "memory_governance_delete": { - let deleted = 0; - for (const id of v.memoryIds || []) { - const existing = await kvInstance.get("mem:memories", id); - if (existing) { - await kvInstance.delete("mem:memories", id); - deleted++; - } - } - kvInstance.persist(); - return textResponse({ - deleted, - requested: (v.memoryIds || []).length, - reason: v.reason, - }); - } - - case "memory_export": { - const memories = await kvInstance.list("mem:memories"); - const sessions = await kvInstance.list("mem:sessions"); - return textResponse({ version: VERSION, memories, sessions }, true); - } - - case "memory_audit": { - const entries = await kvInstance.list("mem:audit"); - const limit = v.limit ?? 50; - return textResponse( - { - entries: (entries as Array>).slice(0, limit), - }, - true, - ); - } - - default: - throw new Error(`Unknown tool: ${v.tool}`); - } -} - -async function handleProxyGeneric( - toolName: string, - args: Record, - handle: ProxyHandle, -): Promise<{ content: Array<{ type: string; text: string }> }> { - // Forward to the server's full MCP surface so non-Claude clients can - // reach all 53 tools (lessons, sentinels, slots, signals, graph, …) - // instead of being capped at the 7 IMPLEMENTED_TOOLS set baked into - // this shim. The server validates arguments per tool. - const result = (await handle.call("/agentmemory/mcp/call", { - method: "POST", - body: JSON.stringify({ name: toolName, arguments: args }), - })) as { content?: Array<{ type: string; text: string }> } | null; - if (result && Array.isArray(result.content)) { - return { content: result.content }; - } - return textResponse(result, true); -} export async function handleToolCall( toolName: string, args: Record, kvInstance: InMemoryKV = kv, ): Promise<{ content: Array<{ type: string; text: string }> }> { - const handle = await resolveHandle(); - announceMode(handle); - - // Tools the local InMemoryKV fallback doesn't implement: forward straight - // to the server. Local validation would otherwise raise "Unknown tool" - // (issue #234). - if (!IMPLEMENTED_TOOLS.has(toolName)) { - if (handle.mode === "proxy") { - try { - return await handleProxyGeneric(toolName, args, handle); - } catch (err) { - process.stderr.write( - `[@agentmemory/mcp] proxy call failed for ${toolName}: ${err instanceof Error ? err.message : String(err)}\n`, - ); - invalidateHandle(); - throw err; - } - } - throw new Error( - `Unknown tool: ${toolName} (local fallback supports only ${[...IMPLEMENTED_TOOLS].join(", ")}; start an agentmemory server and set AGENTMEMORY_URL to use the full tool set)`, - ); - } - - const validated = validate(toolName, args); - if (handle.mode === "proxy") { - try { - return await handleProxy(validated, handle); - } catch (err) { - process.stderr.write( - `[@agentmemory/mcp] proxy call failed for ${toolName}: ${err instanceof Error ? err.message : String(err)}; invalidating handle and falling back to local KV\n`, - ); - invalidateHandle(); - } - } - return handleLocal(validated, kvInstance); + return handleToolCallImpl(toolName, args, null, kvInstance); } export async function handleToolsList(): Promise<{ tools: unknown[] }> { - const debug = process.env["AGENTMEMORY_DEBUG"] === "1" || process.env["AGENTMEMORY_DEBUG"] === "true"; - const handle = await resolveHandle(); - announceMode(handle); - if (debug) { - process.stderr.write( - `[@agentmemory/mcp] tools/list: handle.mode=${handle.mode}${handle.mode === "proxy" ? ` baseUrl=${handle.baseUrl}` : ""}\n`, - ); - } - if (handle.mode === "proxy") { - try { - const remote = (await handle.call("/agentmemory/mcp/tools", { - method: "GET", - })) as { tools?: unknown } | null; - if (debug) { - const shape = remote === null - ? "null" - : typeof remote !== "object" - ? typeof remote - : `keys=${Object.keys(remote as object).join(",")} toolsType=${Array.isArray((remote as { tools?: unknown }).tools) ? `array(len=${((remote as { tools: unknown[] }).tools).length})` : typeof (remote as { tools?: unknown }).tools}`; - process.stderr.write( - `[@agentmemory/mcp] tools/list: remote response shape: ${shape}\n`, - ); - } - if (remote && Array.isArray(remote.tools)) { - if (debug) { - process.stderr.write( - `[@agentmemory/mcp] tools/list: returning ${remote.tools.length} tools from server\n`, - ); - } - return { tools: remote.tools }; - } - process.stderr.write( - `[@agentmemory/mcp] tools/list: server returned unexpected shape (no .tools array); falling back to local IMPLEMENTED_TOOLS list. Set AGENTMEMORY_DEBUG=1 to inspect response.\n`, - ); - } catch (err) { - process.stderr.write( - `[@agentmemory/mcp] tools/list proxy failed: ${err instanceof Error ? err.message : String(err)}; falling back to local list\n`, - ); - invalidateHandle(); - } - } - const fallback = getAllTools().filter((t) => IMPLEMENTED_TOOLS.has(t.name)); - if (debug) { - process.stderr.write( - `[@agentmemory/mcp] tools/list: returning ${fallback.length} local fallback tools (${fallback.map((t) => t.name).join(",")})\n`, - ); - } - return { tools: fallback }; + return handleToolsListImpl(null, kv); } const transport = createStdioTransport(async (method, params) => { diff --git a/src/mcp/stream-http.ts b/src/mcp/stream-http.ts new file mode 100644 index 000000000..1fc6a3b7c --- /dev/null +++ b/src/mcp/stream-http.ts @@ -0,0 +1,331 @@ +/** + * MCP Streamable HTTP transport — manual JSON-RPC implementation. + * + * We hand-roll JSON-RPC routing here instead of using the official + * `@modelcontextprotocol/sdk`'s `NodeStreamableHTTPServerTransport` because + * `handleRequest()` depends on `@hono/node-server`'s `getRequestListener()`, + * which has a known body-parsing bug: it may return an empty or truncated body + * for requests with `Transfer-Encoding: chunked` (common from MCP clients). + * + * Our manual approach gives us full control over body parsing, request + * validation, and error formatting while still complying with the MCP + * Streamable HTTP contract (protocolVersion "2025-03-26"). + * + * Observed in @modelcontextprotocol/node v2.0.0-alpha.2 with chunked transfer encoding. + */ + +import { createServer } from "node:http"; +import type { IncomingMessage, ServerResponse, Server } from "node:http"; +import { randomUUID } from "node:crypto"; +import { timingSafeCompare } from "../auth.js"; +import { handleToolsList, handleToolCall } from "./handler.js"; +import { VERSION } from "../version.js"; +import type { ISdk } from "iii-sdk"; +import type { InMemoryKV } from "./in-memory-kv.js"; + +const MAX_BODY_BYTES = 5 * 1024 * 1024; // 5 MB + +interface JsonRpcMessage { + jsonrpc: string; + id?: string | number | null; + method?: string; + params?: Record; + result?: unknown; + error?: { code: number; message: string; data?: unknown }; +} + +interface SessionEntry { + sessionId: string; + createdAt: Date; + lastUsedAt: Date; +} + +function readBody(req: IncomingMessage): Promise { + return new Promise((resolve, reject) => { + const chunks: Buffer[] = []; + let total = 0; + req.on("data", (chunk: Buffer) => { + const buf = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); + total += buf.length; + if (total > MAX_BODY_BYTES) { + req.destroy(); + reject(new BodyTooLargeError()); + return; + } + chunks.push(buf); + }); + req.on("end", () => resolve(Buffer.concat(chunks).toString("utf-8"))); + req.on("error", reject); + }); +} + +class BodyTooLargeError extends Error { + constructor() { + super("Payload Too Large"); + this.name = "BodyTooLargeError"; + } +} + +function sendJson( + res: ServerResponse, + status: number, + body: unknown, + extraHeaders?: Record, +): void { + const json = JSON.stringify(body); + const headers: Record = { + "Content-Type": "application/json", + "Content-Length": String(Buffer.byteLength(json)), + ...extraHeaders, + }; + res.writeHead(status, headers); + res.end(json); +} + +function errorResponse( + id: string | number | null, + code: number, + message: string, +): JsonRpcMessage { + return { + jsonrpc: "2.0", + id, + error: { code, message }, + }; +} + +/** + * Start the MCP Streamable HTTP server. + * + * Returns the http.Server, the sessions map, and a `shutdown` function. + * Call `shutdown()` to close the server and clear the eviction timer. + */ +export async function startMcpStreamServer( + port: number, + sdk: ISdk | null, + kv: InMemoryKV, + secret?: string, +): Promise<{ + server: Server; + transports: Map; + shutdown: () => Promise; +}> { + const sessions = new Map(); + + // --- periodic eviction of idle sessions (every 10 minutes) --- + const SESSION_IDLE_TTL_MS = 60 * 60 * 1000; // 1 hour + const evictionTimer = setInterval(() => { + const cutoff = Date.now() - SESSION_IDLE_TTL_MS; + for (const [id, entry] of sessions) { + if (entry.lastUsedAt.getTime() < cutoff) { + sessions.delete(id); + } + } + }, 10 * 60 * 1000).unref(); + + const server = createServer(async (req, res) => { + try { + // --- HTTP method validation --- + if (req.method !== "POST" && req.method !== "DELETE") { + res.writeHead(405, { Allow: "POST, DELETE" }); + res.end(JSON.stringify(errorResponse(null, -32000, "Method Not Allowed"))); + return; + } + + // --- Content-Type validation --- + if (req.method === "POST") { + const contentType = (req.headers["content-type"] || "").split(";")[0].trim(); + if (contentType !== "application/json") { + sendJson(res, 415, errorResponse(null, -32000, "Unsupported Media Type: expected application/json")); + return; + } + } + + if (secret) { + const auth = + req.headers["authorization"] || req.headers["Authorization"]; + if ( + typeof auth !== "string" || + !timingSafeCompare(auth, `Bearer ${secret}`) + ) { + sendJson(res, 401, errorResponse(null, -32001, "Unauthorized")); + return; + } + } + + if (req.method === "DELETE") { + const sessionId = req.headers["mcp-session-id"] as string | undefined; + if (sessionId) { + sessions.delete(sessionId); + } + res.writeHead(200); + res.end(); + return; + } + + let bodyStr: string; + try { + bodyStr = await readBody(req); + } catch (err) { + if (err instanceof BodyTooLargeError) { + // Safety guard: readBody() always rejects before any response is sent, + // so headersSent is always false here under normal flow. + if (!res.headersSent) { + res.writeHead(413, { "Content-Type": "text/plain" }); + res.end("Payload Too Large"); + } + } else { + sendJson(res, 500, errorResponse(null, -32603, "Internal error")); + } + return; + } + + let msg: JsonRpcMessage; + try { + msg = JSON.parse(bodyStr); + } catch { + sendJson(res, 400, errorResponse(null, -32700, "Parse error")); + return; + } + + if (!msg || typeof msg.jsonrpc !== "string") { + sendJson(res, 400, errorResponse(null, -32600, "Invalid Request")); + return; + } + + const sessionIdHeader = req.headers["mcp-session-id"] as string | undefined; + + const isInitialize = msg.method === "initialize"; + const isNotification = + msg.id === undefined || msg.id === null; + + if (isNotification) { + res.writeHead(202); + res.end(); + return; + } + + if (!isInitialize && !sessionIdHeader) { + sendJson( + res, + 400, + errorResponse(msg.id ?? null, -32000, "Bad Request: Mcp-Session-Id header is required"), + ); + return; + } + + if (!isInitialize && sessionIdHeader && !sessions.has(sessionIdHeader)) { + sendJson( + res, + 400, + errorResponse(msg.id ?? null, -32000, "Session not found"), + ); + return; + } + + // Bump last-used timestamp for valid session + if (sessionIdHeader && sessions.has(sessionIdHeader)) { + sessions.get(sessionIdHeader)!.lastUsedAt = new Date(); + } + + switch (msg.method) { + case "initialize": { + const sessionId = randomUUID(); + const now = new Date(); + const response: JsonRpcMessage = { + jsonrpc: "2.0", + id: msg.id ?? null, + result: { + protocolVersion: "2025-03-26", + capabilities: { tools: {} }, + serverInfo: { + name: "agentmemory", + version: VERSION, + }, + }, + }; + sessions.set(sessionId, { sessionId, createdAt: now, lastUsedAt: now }); + sendJson(res, 200, response, { "Mcp-Session-Id": sessionId }); + break; + } + + case "tools/list": { + const result = await handleToolsList(sdk, kv); + sendJson(res, 200, { + jsonrpc: "2.0", + id: msg.id ?? null, + result, + }); + break; + } + + case "tools/call": { + const params = msg.params; + if (!params || typeof params.name !== "string") { + sendJson( + res, + 200, + errorResponse(msg.id ?? null, -32602, "Invalid params: name is required"), + ); + return; + } + const toolName = params.name as string; + const args = (params.arguments as Record) || {}; + try { + const result = await handleToolCall(toolName, args, sdk, kv); + sendJson(res, 200, { + jsonrpc: "2.0", + id: msg.id ?? null, + result, + }); + } catch (err) { + sendJson( + res, + 200, + errorResponse( + msg.id ?? null, + -32603, + err instanceof Error ? err.message : "Internal error", + ), + ); + } + break; + } + + default: + sendJson( + res, + 200, + errorResponse( + msg.id ?? null, + -32601, + `Method not found: ${msg.method || "unknown"}`, + ), + ); + break; + } + } catch (err) { + if (!res.headersSent) { + sendJson(res, 500, { + jsonrpc: "2.0", + id: null, + error: { + code: -32603, + message: err instanceof Error ? err.message : "Internal error", + }, + }); + } + } + }); + + await new Promise((resolve) => server.listen(port, "127.0.0.1", resolve)); + + const shutdown = async (): Promise => { + clearInterval(evictionTimer); + await new Promise((resolve, reject) => { + server.close((err) => (err ? reject(err) : resolve())); + }); + }; + + return { server, transports: sessions, shutdown }; +} diff --git a/src/types.ts b/src/types.ts index 6797dfaf9..d27ca6b48 100644 --- a/src/types.ts +++ b/src/types.ts @@ -160,6 +160,8 @@ export interface AgentMemoryConfig { engineUrl: string; restPort: number; streamsPort: number; + /** MCP Streamable HTTP port — defaults to restPort + 3 (3114) */ + mcpPort: number; provider: ProviderConfig; tokenBudget: number; maxObservationsPerSession: number; diff --git a/test/mcp-stream-http.test.ts b/test/mcp-stream-http.test.ts new file mode 100644 index 000000000..42144cac3 --- /dev/null +++ b/test/mcp-stream-http.test.ts @@ -0,0 +1,937 @@ +import { describe, it, expect, beforeEach, afterEach, vi } from "vitest"; +import * as http from "node:http"; + +vi.mock("../src/mcp/rest-proxy.js", () => ({ + resolveHandle: vi.fn(async () => ({ mode: "local", kv: new (await import("../src/mcp/in-memory-kv.js")).InMemoryKV() })), + invalidateHandle: vi.fn(), + setLivezProbe: vi.fn(), + resetHandleForTests: vi.fn(), +})); + +vi.mock("../src/logger.js", () => ({ + logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + bootLog: vi.fn(), +})); + +import { InMemoryKV } from "../src/mcp/in-memory-kv.js"; +import { startMcpStreamServer } from "../src/mcp/stream-http.js"; +import type { Server } from "node:http"; + +const MCP_ACCEPT = "application/json, text/event-stream"; + +function jsonBody(data: unknown): string { + return JSON.stringify(data); +} + +async function httpPost( + port: number, + body: unknown, + headers: Record = {}, +): Promise<{ status: number; headers: http.IncomingHttpHeaders; body: unknown }> { + return new Promise((resolve, reject) => { + const bodyStr = typeof body === "string" ? body : jsonBody(body); + const opts: http.RequestOptions = { + hostname: "127.0.0.1", + port, + method: "POST", + headers: { + "Content-Type": "application/json", + "Accept": MCP_ACCEPT, + "Content-Length": String(Buffer.byteLength(bodyStr)), + ...headers, + }, + }; + const req = http.request(opts, (res) => { + let data = ""; + res.on("data", (chunk) => (data += chunk)); + res.on("end", () => { + try { + const parsed = data.length > 0 ? JSON.parse(data) : null; + resolve({ + status: res.statusCode ?? 0, + headers: res.headers, + body: parsed, + }); + } catch { + resolve({ + status: res.statusCode ?? 0, + headers: res.headers, + body: data, + }); + } + }); + }); + req.on("error", reject); + req.write(bodyStr); + req.end(); + }); +} + +function freePort(): Promise { + return new Promise((resolve, reject) => { + const srv = http.createServer(); + srv.listen(0, "127.0.0.1", () => { + const addr = srv.address(); + const port = typeof addr === "object" && addr ? addr.port : 0; + srv.close(() => resolve(port)); + }); + srv.on("error", reject); + }); +} + +describe("MCP Stream HTTP (T010-T015)", () => { + let server: Server; + let httpPort: number; + let transports: Map; + + beforeEach(async () => { + vi.clearAllMocks(); + }); + + afterEach(async () => { + if (server) { + await new Promise((resolve) => server.close(() => resolve())); + } + }); + + async function startServer(secret?: string) { + httpPort = await freePort(); + const kv = new InMemoryKV(); + const mockSdk = { + trigger: vi.fn(async () => ({})), + registerFunction: vi.fn(), + registerTrigger: vi.fn(), + shutdown: vi.fn(), + }; + const result = await startMcpStreamServer(httpPort, mockSdk as any, kv as any, secret); + server = result.server as unknown as Server; + transports = result.transports as unknown as Map; + } + + // T010 + describe("initialize over HTTP", () => { + it("returns protocolVersion, capabilities, serverInfo, and Mcp-Session-Id header", async () => { + await startServer(); + const res = await httpPost(httpPort, { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { + protocolVersion: "2025-03-26", + capabilities: {}, + clientInfo: { name: "test-client", version: "1.0" }, + }, + }); + + expect(res.status).toBe(200); + const body = res.body as Record; + expect(body.jsonrpc).toBe("2.0"); + expect(body.id).toBe(1); + expect(body.result).toBeDefined(); + expect((body.result as Record).protocolVersion).toBe("2025-03-26"); + expect((body.result as Record).capabilities).toEqual({ tools: {} }); + expect((body.result as Record).serverInfo).toBeDefined(); + const serverInfo = (body.result as Record).serverInfo as Record; + expect(serverInfo.name).toBe("agentmemory"); + expect(typeof serverInfo.version).toBe("string"); + expect(res.headers["mcp-session-id"]).toBeDefined(); + expect(typeof res.headers["mcp-session-id"]).toBe("string"); + }); + + it("rejects initialize without protocolVersion gracefully", async () => { + await startServer(); + const res = await httpPost(httpPort, { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: {}, + }); + + expect(res.status).toBe(200); + const body = res.body as Record; + expect(body.result).toBeDefined(); + }); + + it("handles multiple concurrent initialize requests generating distinct sessions", async () => { + await startServer(); + + const [r1, r2] = await Promise.all([ + httpPost(httpPort, { jsonrpc: "2.0", id: 1, method: "initialize", params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "a", version: "1" } } }), + httpPost(httpPort, { jsonrpc: "2.0", id: 1, method: "initialize", params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "b", version: "1" } } }), + ]); + + expect(r1.status).toBe(200); + expect(r2.status).toBe(200); + expect(r1.headers["mcp-session-id"]).toBeDefined(); + expect(r2.headers["mcp-session-id"]).toBeDefined(); + expect(r1.headers["mcp-session-id"]).not.toBe(r2.headers["mcp-session-id"]); + }); + }); + + // T011 + describe("tools/list over HTTP", () => { + it("returns tools list with correct count when AGENTMEMORY_TOOLS=all", async () => { + process.env["AGENTMEMORY_TOOLS"] = "all"; + await startServer(); + + const initRes = await httpPost(httpPort, { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "test", version: "1.0" } }, + }); + const sessionId = initRes.headers["mcp-session-id"] as string; + expect(sessionId).toBeDefined(); + + const res = await httpPost( + httpPort, + { jsonrpc: "2.0", id: 2, method: "tools/list", params: {} }, + { "Mcp-Session-Id": sessionId }, + ); + + expect(res.status).toBe(200); + const body = res.body as Record; + expect(body.result).toBeDefined(); + const result = body.result as Record; + expect(result.tools).toBeDefined(); + expect(Array.isArray(result.tools)).toBe(true); + const tools = result.tools as unknown[]; + // In test mode (local fallback), IMPLEMENTED_TOOLS returns 7 tools + // (production connected to engine returns all 53) + expect(tools.length).toBeGreaterThanOrEqual(7); + }); + + it("returns reduced tool set with AGENTMEMORY_TOOLS=core", async () => { + process.env["AGENTMEMORY_TOOLS"] = "core"; + await startServer(); + + const initRes = await httpPost(httpPort, { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "test", version: "1.0" } }, + }); + const sessionId = initRes.headers["mcp-session-id"] as string; + + const res = await httpPost( + httpPort, + { jsonrpc: "2.0", id: 2, method: "tools/list", params: {} }, + { "Mcp-Session-Id": sessionId }, + ); + + const body = res.body as Record; + const result = body.result as Record; + const tools = result.tools as unknown[]; + // In test mode, IMPLEMENTED_TOOLS filter applies (7 tools) + expect(tools.length).toBeGreaterThan(0); + for (const tool of tools) { + const t = tool as Record; + expect(t.name).toBeDefined(); + expect(t.description).toBeDefined(); + expect(t.inputSchema).toBeDefined(); + } + }); + }); + + // T012 + describe("tools/call over HTTP", () => { + it("forwards tool call and returns content array", async () => { + process.env["AGENTMEMORY_TOOLS"] = "all"; + await startServer(); + + const initRes = await httpPost(httpPort, { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "test", version: "1.0" } }, + }); + const sessionId = initRes.headers["mcp-session-id"] as string; + + const res = await httpPost( + httpPort, + { + jsonrpc: "2.0", + id: 3, + method: "tools/call", + params: { name: "memory_export", arguments: {} }, + }, + { "Mcp-Session-Id": sessionId }, + ); + + expect(res.status).toBe(200); + const body = res.body as Record; + expect(body.result).toBeDefined(); + const result = body.result as Record; + expect(result.content).toBeDefined(); + expect(Array.isArray(result.content)).toBe(true); + const content = result.content as Array<{ type: string; text: string }>; + expect(content.length).toBeGreaterThan(0); + expect(content[0].type).toBe("text"); + expect(typeof content[0].text).toBe("string"); + }); + + it("returns error for unknown tool", async () => { + process.env["AGENTMEMORY_TOOLS"] = "all"; + await startServer(); + + const initRes = await httpPost(httpPort, { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "test", version: "1.0" } }, + }); + const sessionId = initRes.headers["mcp-session-id"] as string; + + const res = await httpPost( + httpPort, + { + jsonrpc: "2.0", + id: 4, + method: "tools/call", + params: { name: "nonexistent_tool", arguments: {} }, + }, + { "Mcp-Session-Id": sessionId }, + ); + + expect(res.status).toBe(200); + const body = res.body as Record; + expect(body.error).toBeDefined(); + }); + + it("returns error for missing tool name", async () => { + process.env["AGENTMEMORY_TOOLS"] = "all"; + await startServer(); + + const initRes = await httpPost(httpPort, { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "test", version: "1.0" } }, + }); + const sessionId = initRes.headers["mcp-session-id"] as string; + + const res = await httpPost( + httpPort, + { jsonrpc: "2.0", id: 5, method: "tools/call", params: {} }, + { "Mcp-Session-Id": sessionId }, + ); + + expect(res.status).toBe(200); + const body = res.body as Record; + expect(body.error).toBeDefined(); + }); + }); + + // T013 + describe("Bearer auth", () => { + it("returns 401 without token when secret is set", async () => { + await startServer("test-secret"); + const res = await httpPost(httpPort, { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "test", version: "1.0" } }, + }); + + expect(res.status).toBe(401); + const body = res.body as Record; + expect(body.error).toBeDefined(); + }); + + it("returns 401 with wrong token", async () => { + await startServer("test-secret"); + const res = await httpPost( + httpPort, + { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "test", version: "1.0" } }, + }, + { Authorization: "Bearer wrong-token" }, + ); + + expect(res.status).toBe(401); + }); + + it("returns 200 with correct token", async () => { + await startServer("test-secret"); + const res = await httpPost( + httpPort, + { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "test", version: "1.0" } }, + }, + { Authorization: "Bearer test-secret" }, + ); + + expect(res.status).toBe(200); + const body = res.body as Record; + expect(body.result).toBeDefined(); + }); + + it("skips auth when secret is undefined", async () => { + await startServer(undefined); + const res = await httpPost(httpPort, { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "test", version: "1.0" } }, + }); + + expect(res.status).toBe(200); + const body = res.body as Record; + expect(body.result).toBeDefined(); + }); + + it("skips auth when secret is empty string", async () => { + await startServer(""); + const res = await httpPost(httpPort, { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "test", version: "1.0" } }, + }); + + expect(res.status).toBe(200); + }); + }); + + // T014 + describe("JSON-RPC errors", () => { + it("returns parse error (-32700) for invalid JSON", async () => { + await startServer(); + const bodyStr = "not-valid-json{{{"; + const res = await new Promise<{ status: number; body: unknown }>((resolve, reject) => { + const req = http.request( + { + hostname: "127.0.0.1", + port: httpPort, + method: "POST", + headers: { + "Content-Type": "application/json", + "Accept": MCP_ACCEPT, + "Content-Length": String(Buffer.byteLength(bodyStr)), + }, + }, + (res) => { + let data = ""; + res.on("data", (chunk) => (data += chunk)); + res.on("end", () => { + try { + resolve({ status: res.statusCode ?? 0, body: JSON.parse(data) }); + } catch { + resolve({ status: res.statusCode ?? 0, body: data }); + } + }); + }, + ); + req.on("error", reject); + req.write(bodyStr); + req.end(); + }); + + expect(res.status).toBe(400); + const body = res.body as Record; + expect(body.error).toBeDefined(); + expect((body.error as Record).code).toBe(-32700); + }); + + it("returns method not found (-32601) for unknown methods", async () => { + await startServer(); + + // Initialize first to get a session + const initRes = await httpPost(httpPort, { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "test", version: "1.0" } }, + }); + const sessionId = initRes.headers["mcp-session-id"] as string; + + const res = await httpPost( + httpPort, + { + jsonrpc: "2.0", + id: 2, + method: "unknown/method", + params: {}, + }, + { "Mcp-Session-Id": sessionId }, + ); + + const body = res.body as Record; + expect(body.error).toBeDefined(); + expect((body.error as Record).code).toBe(-32601); + }); + + it("returns invalid params (-32602) for tools/call without params", async () => { + process.env["AGENTMEMORY_TOOLS"] = "all"; + await startServer(); + + const initRes = await httpPost(httpPort, { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "test", version: "1.0" } }, + }); + const sessionId = initRes.headers["mcp-session-id"] as string; + + const res = await httpPost( + httpPort, + { jsonrpc: "2.0", id: 12, method: "tools/call" }, + { "Mcp-Session-Id": sessionId }, + ); + + const body = res.body as Record; + expect(body.error).toBeDefined(); + expect((body.error as Record).code).toBe(-32602); + }); + }); + + // T015 + describe("session management", () => { + it("initialize creates a session with Mcp-Session-Id header", async () => { + await startServer(); + + const res = await httpPost(httpPort, { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "test", version: "1.0" } }, + }); + + expect(res.status).toBe(200); + expect(res.headers["mcp-session-id"]).toBeDefined(); + const sessionId = res.headers["mcp-session-id"] as string; + expect(typeof sessionId).toBe("string"); + expect(sessionId.length).toBeGreaterThan(0); + }); + + it("reuses session via Mcp-Session-Id header for tools/list", async () => { + await startServer(); + + const initRes = await httpPost(httpPort, { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "test", version: "1.0" } }, + }); + const sessionId = initRes.headers["mcp-session-id"] as string; + + const toolsRes = await httpPost( + httpPort, + { jsonrpc: "2.0", id: 2, method: "tools/list", params: {} }, + { "Mcp-Session-Id": sessionId }, + ); + + expect(toolsRes.status).toBe(200); + const body = toolsRes.body as Record; + expect(body.result).toBeDefined(); + }); + + it("rejects invalid session ID with error code -32000", async () => { + await startServer(); + + const res = await httpPost( + httpPort, + { jsonrpc: "2.0", id: 1, method: "tools/list", params: {} }, + { "Mcp-Session-Id": "invalid-nonexistent-session-id" }, + ); + + expect([400, 404]).toContain(res.status); + const body = res.body as Record; + if (body.error) { + expect((body.error as Record).code).toBe(-32000); + } + }); + + it("returns 400 for non-initialize requests without session ID", async () => { + await startServer(); + + const res = await httpPost(httpPort, { + jsonrpc: "2.0", + id: 1, + method: "tools/list", + params: {}, + }); + + expect(res.status).toBe(400); + }); + + it("multiple initialize creates separate sessions", async () => { + await startServer(); + + const r1 = await httpPost(httpPort, { + jsonrpc: "2.0", id: 1, method: "initialize", + params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "a", version: "1" } }, + }); + const r2 = await httpPost(httpPort, { + jsonrpc: "2.0", id: 1, method: "initialize", + params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "b", version: "1" } }, + }); + + const s1 = r1.headers["mcp-session-id"] as string; + const s2 = r2.headers["mcp-session-id"] as string; + + expect(s1).toBeDefined(); + expect(s2).toBeDefined(); + expect(s1).not.toBe(s2); + + // Both sessions should work independently + const t1 = await httpPost( + httpPort, + { jsonrpc: "2.0", id: 2, method: "tools/list", params: {} }, + { "Mcp-Session-Id": s1 }, + ); + const t2 = await httpPost( + httpPort, + { jsonrpc: "2.0", id: 2, method: "tools/list", params: {} }, + { "Mcp-Session-Id": s2 }, + ); + + expect(t1.status).toBe(200); + expect(t2.status).toBe(200); + }); + }); + + // T017 + describe("notification handling (T017)", () => { + it("handles initialized notification without response", async () => { + await startServer(); + + const initRes = await httpPost(httpPort, { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "test", version: "1.0" } }, + }); + const sessionId = initRes.headers["mcp-session-id"] as string; + + // Send initialized notification (no id field) + const notifRes = await httpPost( + httpPort, + { + jsonrpc: "2.0", + method: "notifications/initialized", + params: {}, + }, + { "Mcp-Session-Id": sessionId }, + ); + + // Notifications should be accepted (200 or 202) + expect([200, 202]).toContain(notifRes.status); + }); + }); + + // T019 — HTTP request validation + describe("HTTP method validation", () => { + it("returns 405 for GET requests with Allow header", async () => { + await startServer(); + const res = await new Promise<{ status: number; headers: http.IncomingHttpHeaders; body: string }>((resolve, reject) => { + const req = http.get(`http://127.0.0.1:${httpPort}/`, (res) => { + let data = ""; + res.on("data", (chunk) => (data += chunk)); + res.on("end", () => resolve({ status: res.statusCode ?? 0, headers: res.headers, body: data })); + }); + req.on("error", reject); + }); + expect(res.status).toBe(405); + expect(res.headers["allow"]).toBe("POST, DELETE"); + }); + + it("returns 405 for OPTIONS requests", async () => { + await startServer(); + const res = await new Promise<{ status: number; body: unknown }>((resolve, reject) => { + const req = http.request( + { hostname: "127.0.0.1", port: httpPort, method: "OPTIONS" }, + (res) => { + let data = ""; + res.on("data", (chunk) => (data += chunk)); + res.on("end", () => { + try { + resolve({ status: res.statusCode ?? 0, body: JSON.parse(data) }); + } catch { + resolve({ status: res.statusCode ?? 0, body: data }); + } + }); + }, + ); + req.on("error", reject); + req.end(); + }); + expect(res.status).toBe(405); + }); + + it("returns 405 for HEAD requests", async () => { + await startServer(); + const res = await new Promise<{ status: number }>((resolve, reject) => { + const req = http.request( + { hostname: "127.0.0.1", port: httpPort, method: "HEAD" }, + (res) => { + res.resume(); + resolve({ status: res.statusCode ?? 0 }); + }, + ); + req.on("error", reject); + req.end(); + }); + expect(res.status).toBe(405); + }); + }); + + describe("Content-Type validation", () => { + it("returns 415 for text/plain content type", async () => { + await startServer(); + const bodyStr = JSON.stringify({ jsonrpc: "2.0", id: 1, method: "initialize", params: {} }); + const res = await new Promise<{ status: number; body: unknown }>((resolve, reject) => { + const req = http.request( + { + hostname: "127.0.0.1", + port: httpPort, + method: "POST", + headers: { + "Content-Type": "text/plain", + "Content-Length": String(Buffer.byteLength(bodyStr)), + }, + }, + (res) => { + let data = ""; + res.on("data", (chunk) => (data += chunk)); + res.on("end", () => { + try { + resolve({ status: res.statusCode ?? 0, body: JSON.parse(data) }); + } catch { + resolve({ status: res.statusCode ?? 0, body: data }); + } + }); + }, + ); + req.on("error", reject); + req.write(bodyStr); + req.end(); + }); + expect(res.status).toBe(415); + }); + + it("returns 415 for multipart/form-data content type", async () => { + await startServer(); + const res = await new Promise<{ status: number; body: unknown }>((resolve, reject) => { + const req = http.request( + { + hostname: "127.0.0.1", + port: httpPort, + method: "POST", + headers: { "Content-Type": "multipart/form-data", "Content-Length": "0" }, + }, + (res) => { + let data = ""; + res.on("data", (chunk) => (data += chunk)); + res.on("end", () => { + try { + resolve({ status: res.statusCode ?? 0, body: JSON.parse(data) }); + } catch { + resolve({ status: res.statusCode ?? 0, body: data }); + } + }); + }, + ); + req.on("error", reject); + req.end(); + }); + expect(res.status).toBe(415); + }); + + it("returns 415 for missing Content-Type", async () => { + await startServer(); + const bodyStr = JSON.stringify({ jsonrpc: "2.0", id: 1, method: "initialize", params: {} }); + const res = await new Promise<{ status: number; body: unknown }>((resolve, reject) => { + const req = http.request( + { + hostname: "127.0.0.1", + port: httpPort, + method: "POST", + headers: { "Content-Length": String(Buffer.byteLength(bodyStr)) }, + }, + (res) => { + let data = ""; + res.on("data", (chunk) => (data += chunk)); + res.on("end", () => { + try { + resolve({ status: res.statusCode ?? 0, body: JSON.parse(data) }); + } catch { + resolve({ status: res.statusCode ?? 0, body: data }); + } + }); + }, + ); + req.on("error", reject); + req.write(bodyStr); + req.end(); + }); + expect(res.status).toBe(415); + }); + + it("accepts application/json with charset parameter", async () => { + await startServer(); + const res = await httpPost(httpPort, { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "test", version: "1.0" } }, + }); + // httpPost helper always sets Content-Type: application/json + expect(res.status).toBe(200); + }); + }); + + describe("body size limit", () => { + it("returns 413 for payload exceeding 5MB", async () => { + await startServer(); + // Create a payload larger than 5MB + const largeStr = "x".repeat(6 * 1024 * 1024); // 6 MB + const bodyStr = JSON.stringify({ jsonrpc: "2.0", id: 1, method: "initialize", params: { data: largeStr } }); + const res = await new Promise<{ status: number }>((resolve, reject) => { + const req = http.request( + { + hostname: "127.0.0.1", + port: httpPort, + method: "POST", + headers: { + "Content-Type": "application/json", + "Content-Length": String(Buffer.byteLength(bodyStr)), + }, + }, + (res) => { + res.resume(); + resolve({ status: res.statusCode ?? 0 }); + }, + ); + req.on("error", (err) => { + // The server may destroy the connection on oversize + resolve({ status: 413 }); + }); + req.write(bodyStr); + req.end(); + }); + expect(res.status).toBe(413); + }); + + it("accepts payload under 5MB", async () => { + await startServer(); + const res = await httpPost(httpPort, { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "test", version: "1.0" } }, + }); + expect(res.status).toBe(200); + }); + }); + + describe("DELETE request handling", () => { + it("returns 200 for DELETE requests", async () => { + await startServer(); + const res = await new Promise<{ status: number }>((resolve, reject) => { + const req = http.request( + { hostname: "127.0.0.1", port: httpPort, method: "DELETE" }, + (res) => { + res.resume(); + resolve({ status: res.statusCode ?? 0 }); + }, + ); + req.on("error", reject); + req.end(); + }); + expect(res.status).toBe(200); + }); + }); + + describe("session lastUsedAt tracking", () => { + it("updates lastUsedAt on each request with valid session", async () => { + await startServer(); + + const initRes = await httpPost(httpPort, { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "test", version: "1.0" } }, + }); + const sessionId = initRes.headers["mcp-session-id"] as string; + expect(sessionId).toBeDefined(); + + const entry1 = transports.get(sessionId) as { lastUsedAt: Date } | undefined; + expect(entry1).toBeDefined(); + const firstLastUsed = entry1!.lastUsedAt.getTime(); + + // Small delay then another request + await new Promise((r) => setTimeout(r, 50)); + + await httpPost( + httpPort, + { jsonrpc: "2.0", id: 2, method: "tools/list", params: {} }, + { "Mcp-Session-Id": sessionId }, + ); + + const entry2 = transports.get(sessionId) as { lastUsedAt: Date } | undefined; + expect(entry2).toBeDefined(); + expect(entry2!.lastUsedAt.getTime()).toBeGreaterThan(firstLastUsed); + }); + + it("evicts sessions with lastUsedAt older than 1 hour", async () => { + vi.useFakeTimers(); + try { + await startServer(); + + // Create a session normally + const initRes = await httpPost(httpPort, { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "test", version: "1.0" } }, + }); + const sessionId = initRes.headers["mcp-session-id"] as string; + expect(sessionId).toBeDefined(); + expect(transports.has(sessionId)).toBe(true); + + // Manually age the session's lastUsedAt to 2 hours ago + const entry = transports.get(sessionId) as { lastUsedAt: Date } | undefined; + expect(entry).toBeDefined(); + entry!.lastUsedAt = new Date(Date.now() - 2 * 60 * 60 * 1000); + + // Advance time past the eviction interval (10 min + 1ms) + vi.advanceTimersByTime(10 * 60 * 1000 + 1); + + // Session should be evicted + expect(transports.has(sessionId)).toBe(false); + } finally { + vi.useRealTimers(); + } + }); + + it("does not evict recently used sessions", async () => { + vi.useFakeTimers(); + try { + await startServer(); + + const initRes = await httpPost(httpPort, { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "test", version: "1.0" } }, + }); + const sessionId = initRes.headers["mcp-session-id"] as string; + + // Advance time but keep session recent (under 1 hour idle) + vi.advanceTimersByTime(10 * 60 * 1000 + 1); + + // Session should still be alive (lastUsedAt is current) + expect(transports.has(sessionId)).toBe(true); + } finally { + vi.useRealTimers(); + } + }); + }); +}); diff --git a/test/mcp-transport-coexistence.test.ts b/test/mcp-transport-coexistence.test.ts new file mode 100644 index 000000000..869f3cabf --- /dev/null +++ b/test/mcp-transport-coexistence.test.ts @@ -0,0 +1,250 @@ +import { describe, it, expect, beforeEach, afterEach, vi } from "vitest"; +import * as http from "node:http"; + +vi.mock("../src/mcp/rest-proxy.js", () => ({ + resolveHandle: vi.fn(async () => ({ mode: "local", kv: new (await import("../src/mcp/in-memory-kv.js")).InMemoryKV() })), + invalidateHandle: vi.fn(), + setLivezProbe: vi.fn(), + resetHandleForTests: vi.fn(), +})); + +vi.mock("../src/logger.js", () => ({ + logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn() }, + bootLog: vi.fn(), +})); + +import { InMemoryKV } from "../src/mcp/in-memory-kv.js"; +import { startMcpStreamServer } from "../src/mcp/stream-http.js"; +import { handleToolsList } from "../src/mcp/handler.js"; +import type { Server } from "node:http"; + +const MCP_ACCEPT = "application/json, text/event-stream"; + +async function httpPost( + port: number, + body: unknown, + headers: Record = {}, +): Promise<{ status: number; headers: http.IncomingHttpHeaders; body: unknown }> { + return new Promise((resolve, reject) => { + const bodyStr = JSON.stringify(body); + const opts: http.RequestOptions = { + hostname: "127.0.0.1", + port, + method: "POST", + headers: { + "Content-Type": "application/json", + "Accept": MCP_ACCEPT, + "Content-Length": String(Buffer.byteLength(bodyStr)), + ...headers, + }, + }; + const req = http.request(opts, (res) => { + let data = ""; + res.on("data", (chunk) => (data += chunk)); + res.on("end", () => { + try { + resolve({ + status: res.statusCode ?? 0, + headers: res.headers, + body: JSON.parse(data), + }); + } catch { + resolve({ + status: res.statusCode ?? 0, + headers: res.headers, + body: data, + }); + } + }); + }); + req.on("error", reject); + req.write(bodyStr); + req.end(); + }); +} + +function freePort(): Promise { + return new Promise((resolve, reject) => { + const srv = http.createServer(); + srv.listen(0, "127.0.0.1", () => { + const addr = srv.address(); + const port = typeof addr === "object" && addr ? addr.port : 0; + srv.close(() => resolve(port)); + }); + srv.on("error", reject); + }); +} + +describe("MCP Transport Coexistence (T016-T017)", () => { + let server: Server; + let httpPort: number; + let kv: InMemoryKV; + let mockSdk: any; + + beforeEach(async () => { + vi.clearAllMocks(); + process.env["AGENTMEMORY_TOOLS"] = "all"; + kv = new InMemoryKV(); + httpPort = await freePort(); + mockSdk = { + trigger: vi.fn(async () => ({})), + registerFunction: vi.fn(), + registerTrigger: vi.fn(), + shutdown: vi.fn(), + }; + const result = await startMcpStreamServer(httpPort, mockSdk as any, kv as any, undefined); + server = result.server as unknown as Server; + }); + + afterEach(async () => { + if (server) { + await new Promise((resolve) => server.close(() => resolve())); + } + }); + + // T016 + describe("identical tool sets", () => { + it("returns the same tool names via HTTP transport as handleToolsList()", async () => { + const initRes = await httpPost(httpPort, { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "test", version: "1.0" } }, + }); + const sessionId = initRes.headers["mcp-session-id"] as string; + + const toolsRes = await httpPost( + httpPort, + { jsonrpc: "2.0", id: 2, method: "tools/list", params: {} }, + { "Mcp-Session-Id": sessionId }, + ); + + const body = toolsRes.body as Record; + const result = body.result as Record; + const httpTools = (result.tools as Array<{ name: string }>).map((t) => t.name).sort(); + + // Compare against what handleToolsList returns directly + const handlerResult = await handleToolsList(mockSdk, kv); + const handlerTools = (handlerResult.tools as Array<{ name: string }>).map((t) => t.name).sort(); + + expect(httpTools).toEqual(handlerTools); + expect(httpTools.length).toBe(handlerTools.length); + expect(httpTools.length).toBeGreaterThan(0); + }); + + it("returns consistent tool schemas via HTTP matching handleToolsList()", async () => { + const initRes = await httpPost(httpPort, { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "test", version: "1.0" } }, + }); + const sessionId = initRes.headers["mcp-session-id"] as string; + + const toolsRes = await httpPost( + httpPort, + { jsonrpc: "2.0", id: 2, method: "tools/list", params: {} }, + { "Mcp-Session-Id": sessionId }, + ); + + const body = toolsRes.body as Record; + const httpTools = (body.result as Record).tools as Array>; + + const handlerResult = await handleToolsList(mockSdk, kv); + const handlerTools = handlerResult.tools as Array>; + + expect(httpTools.length).toBe(handlerTools.length); + + for (let i = 0; i < handlerTools.length; i++) { + const httpTool = httpTools[i]; + const handlerTool = handlerTools[i]; + expect(httpTool.name).toBe(handlerTool.name); + expect(httpTool.description).toBe(handlerTool.description); + expect(httpTool.inputSchema).toEqual(handlerTool.inputSchema); + } + }); + + it("returns identical results for the same tool call via HTTP transport", async () => { + const initRes = await httpPost(httpPort, { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "test", version: "1.0" } }, + }); + const sessionId = initRes.headers["mcp-session-id"] as string; + + const res = await httpPost( + httpPort, + { + jsonrpc: "2.0", + id: 3, + method: "tools/call", + params: { name: "memory_export", arguments: {} }, + }, + { "Mcp-Session-Id": sessionId }, + ); + + expect(res.status).toBe(200); + const body = res.body as Record; + expect(body.result).toBeDefined(); + const result = body.result as Record; + expect(result.content).toBeDefined(); + expect(Array.isArray(result.content)).toBe(true); + const content = result.content as Array<{ type: string; text: string }>; + expect(content[0].type).toBe("text"); + + const parsed = JSON.parse(content[0].text); + expect(parsed).toBeDefined(); + expect(parsed.version).toBeDefined(); + expect(Array.isArray(parsed.memories)).toBe(true); + expect(Array.isArray(parsed.sessions)).toBe(true); + }); + }); + + // T017 + describe("notification handling", () => { + it("notifications without id field produce no response body", async () => { + const initRes = await httpPost(httpPort, { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "test", version: "1.0" } }, + }); + const sessionId = initRes.headers["mcp-session-id"] as string; + + const res = await httpPost( + httpPort, + { + jsonrpc: "2.0", + method: "notifications/initialized", + params: {}, + }, + { "Mcp-Session-Id": sessionId }, + ); + + expect([200, 202]).toContain(res.status); + }); + + it("server handles JSON-RPC without id field (notification)", async () => { + const initRes = await httpPost(httpPort, { + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "test", version: "1.0" } }, + }); + const sessionId = initRes.headers["mcp-session-id"] as string; + + const res = await httpPost( + httpPort, + { + jsonrpc: "2.0", + method: "notifications/cancelled", + params: { requestId: 1 }, + }, + { "Mcp-Session-Id": sessionId }, + ); + + expect([200, 202, 204]).toContain(res.status); + }); + }); +}); diff --git a/test/multi-instance-port.test.ts b/test/multi-instance-port.test.ts index 4343365b3..a0c5e6e69 100644 --- a/test/multi-instance-port.test.ts +++ b/test/multi-instance-port.test.ts @@ -7,6 +7,7 @@ const PORT_ENVS = [ "III_STREAMS_PORT", "III_ENGINE_PORT", "III_ENGINE_URL", + "AGENTMEMORY_MCP_PORT", ] as const; describe("multi-instance port auto-derive (#750)", () => { @@ -29,35 +30,49 @@ describe("multi-instance port auto-derive (#750)", () => { } }); - it("default REST anchor yields canonical 3111/3112/49134 quartet", () => { + it("default REST anchor yields canonical 3111/3112/3114/49134 family", () => { const cfg = loadConfig(); expect(cfg.restPort).toBe(3111); expect(cfg.streamsPort).toBe(3112); + expect(cfg.mcpPort).toBe(3114); expect(cfg.engineUrl).toBe("ws://localhost:49134"); }); - it("relocating REST drags streams + engine with it", () => { + it("relocating REST drags streams + MCP + engine with it", () => { process.env["III_REST_PORT"] = "3211"; const cfg = loadConfig(); expect(cfg.restPort).toBe(3211); expect(cfg.streamsPort).toBe(3212); + expect(cfg.mcpPort).toBe(3214); expect(cfg.engineUrl).toBe("ws://localhost:49234"); }); - it("instance N=2 block (3311) lands on 3312 + 49334", () => { + it("instance N=2 block (3311) lands on 3312 + 3314 + 49334", () => { process.env["III_REST_PORT"] = "3311"; const cfg = loadConfig(); expect(cfg.restPort).toBe(3311); expect(cfg.streamsPort).toBe(3312); + expect(cfg.mcpPort).toBe(3314); expect(cfg.engineUrl).toBe("ws://localhost:49334"); }); - it("explicit III_STREAM_PORT pins streams without affecting REST/engine", () => { + it("explicit III_STREAM_PORT pins streams without affecting REST/MCP/engine", () => { process.env["III_REST_PORT"] = "3211"; process.env["III_STREAM_PORT"] = "9999"; const cfg = loadConfig(); expect(cfg.restPort).toBe(3211); expect(cfg.streamsPort).toBe(9999); + expect(cfg.mcpPort).toBe(3214); + expect(cfg.engineUrl).toBe("ws://localhost:49234"); + }); + + it("explicit AGENTMEMORY_MCP_PORT pins MCP without affecting REST/streams/engine", () => { + process.env["III_REST_PORT"] = "3211"; + process.env["AGENTMEMORY_MCP_PORT"] = "8888"; + const cfg = loadConfig(); + expect(cfg.restPort).toBe(3211); + expect(cfg.streamsPort).toBe(3212); + expect(cfg.mcpPort).toBe(8888); expect(cfg.engineUrl).toBe("ws://localhost:49234"); });