diff --git a/docs/memory-mcp.md b/docs/memory-mcp.md new file mode 100644 index 00000000..2048c583 --- /dev/null +++ b/docs/memory-mcp.md @@ -0,0 +1,191 @@ +# 专家记忆 MCP Server 使用文档 + +> Octop 将专家记忆通过 **MCP（Streamable HTTP）** 暴露给外部 agent（编码 agent、机器人等）， +> 与进程内 `MemoryService` 能力对齐。外部 agent 可读写专家记忆；每次写入都会带上 +> `source` 标记，召回时可追溯来源。 +> +> 适用版本：`feat/memory-mcp-extension` 分支（9 工具面）。 + +--- + +## 1. 连接信息 + +| 项 | 值 | +|---|---| +| Endpoint | `http://:/mcp/memory` | +| 协议 | MCP Streamable HTTP（SSE + JSON） | +| 鉴权 | `Authorization: Bearer ` | +| 专家绑定 | `X-Octop-Agent-Id: ` 请求头（一连接绑定一专家） | + +### 1.1 鉴权（fail-closed） + +- 服务端通过环境变量 `OCTOP_MEMORY_MCP_TOKEN` 配置独立 token；**未配置时不挂载** `/mcp/memory`（安全默认）。 +- 每个请求必须带 `Authorization: Bearer `，否则返回 `401`。 +- Token 建议与 Octop 登录 JWT 完全隔离（独立凭据，仅限记忆通道使用）。 + +### 1.2 专家绑定（一连接一专家） + +- 端点只有一个 `/mcp/memory`，专家在**连接时**通过 `X-Octop-Agent-Id` 头选择（如 `main`、`MRA7KP`）。 +- 一次连接绑定一个专家，工具调用时**不再**传 agent id——调用方只需在建立会话时固定一个专家。 +- 所有读写都落在该专家的 `Memory` 实例（默认 SQLite；PG 控制面可显式开启）。 + +### 1.3 调用者追溯 + +- 内网增强：MCP 请求的调用者 user id 通过 `X-Octop-User-Id` 头（或工具显式 `user` 参数）传递， + `memory_capture` / `memory_save` / `memory_update` 会按调用者做 per-user 追溯。 +- stateless Streamable HTTP 下 mcp SDK 不提供 `ctx.request_context`，实现用 ContextVar 跨 + ASGI 中间件 → 工具传递，因此每次请求头里的调用者身份是可靠的。 + +--- + +## 2. 工具清单（9 个） + +### 2.1 读取 + +| 工具 | 参数 | 说明 | +|---|---|---| +| `memory_recall` | `query: str`, `limit: int = 5`, `user: str | None` | 语义召回专家记忆（原子/树，L2）。日常使用入口 | +| `memory_raws` | `query: str | None`, `session_id: str | None`, `host: str | None`, `user: str | None`, `limit: int = 50` | 查 L0 原始事件（采集即可见，提取前也能查） | + +### 2.2 写入 + +| 工具 | 参数 | 说明 | +|---|---|---| +| `memory_capture` | `content: str`, `source: str`, `session_id: str | None`, `user: str | None` | 记录一条 **L0 原始事件**，走提取流水线（extract → candidate → promote → atom）。适合记录对话/事件原文 | +| `memory_save` | `content: str`, `source: str`, `topic: str | None`, `user: str | None` | 直接持久化一条**结构化事实**到原子层（L2，不经过提取）。适合已知的明确事实 | +| `memory_update` | `atom_id: str`, `new_content: str`, `source: str`, `note: str = "mcp update"`, `user: str | None` | 显式更新一条记忆：旧 atom 标记 deprecated，新事实立即可召回 | + +### 2.3 提取 / 审核流水线 + +| 工具 | 参数 | 说明 | +|---|---|---| +| `memory_extract` | `session_id: str | None`, `limit: int = 100`, `promote: bool = False` | 手动触发提取：取最近 L0 事件 → LLM 类型化提取 → 候选（pending）。`promote=True` 时直接晋升检查 | +| `memory_candidates` | `status: str | None`, `session_id: str | None`, `limit: int = 50` | 列出候选记忆（默认 pending 队列；可用 status 过滤 promoted / rejected / needs_review） | +| `memory_promote` | `candidate_ids: list[str]`, `importance: str | None` | 审核晋升候选（L1 → L2 原子），可覆盖 importance | +| `memory_reject` | `candidate_id: str`, `reason: str = "rejected by external caller"` | 拒绝候选并记录原因（journal 可审计） | + +--- + +## 3. 使用示例 + +### 3.1 直接 HTTP（curl） + +```bash +TOKEN="" +AGENT="main" # 绑定的专家 +BASE="http://127.0.0.1:8088/mcp/memory" + +# 初始化（MCP 握手） +curl -s -X POST "$BASE/" \ + -H "Authorization: Bearer $TOKEN" \ + -H "X-Octop-Agent-Id: $AGENT" \ + -H "Content-Type: application/json" \ + -H "Accept: application/json, text/event-stream" \ + -d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-03-26","capabilities":{},"clientInfo":{"name":"cli","version":"1.0"}}}' +``` + +### 3.2 记录一条原始事件（L0） + +```json +{ + "jsonrpc": "2.0", + "id": 2, + "method": "tools/call", + "params": { + "name": "memory_capture", + "arguments": { + "content": "user reported: the report panel banner is not rendering", + "source": "review-bot", + "user": "alice" + } + } +} +``` + +### 3.3 直接保存一条事实（L2） + +```json +{ + "jsonrpc": "2.0", + "id": 3, + "method": "tools/call", + "params": { + "name": "memory_save", + "arguments": { + "content": "机票记忆体系 M1 采用 Octop 专家记忆平台作为目标服务", + "source": "planning-agent", + "topic": "flight-memory-system" + } + } +} +``` + +### 3.4 召回 + +```json +{ + "jsonrpc": "2.0", + "id": 4, + "method": "tools/call", + "params": { + "name": "memory_recall", + "arguments": { "query": "机票记忆体系", "limit": 5 } + } +} +``` + +### 3.5 用 MCP 客户端 SDK + +```python +# pip install mcp +import asyncio +from mcp import ClientSession +from mcp.client.streamable_http import streamablehttp_client + +async def main(): + # 注意：endpoint 需以 / 结尾，且须带鉴权与专家绑定头 + async with streamablehttp_client( + "http://127.0.0.1:8088/mcp/memory/", + headers={"Authorization": "Bearer ", + "X-Octop-Agent-Id": "main"}, + ) as (read, write, _): + async with ClientSession(read, write) as session: + await session.initialize() + tools = await session.list_tools() + print([t.name for t in tools.tools]) # 9 个 memory_* 工具 + res = await session.call_tool("memory_recall", {"query": "octop 沙箱"}) + print(res) + +asyncio.run(main()) +``` + +--- + +## 4. 部署与配置 + +| 配置 | 说明 | +|---|---| +| `OCTOP_MEMORY_MCP_TOKEN` | 必填；未设置则 `/mcp/memory` 不挂载（fail-closed） | +| `X-Octop-Agent-Id` | 连接时必填；专家 id（`octop agent list` 可查） | +| 记忆后端 | 默认 SQLite（`~/.octop/agents//memory.sqlite`）；PG 控制面显式开启 | + +### 4.1 开启步骤 + +1. 设置环境变量 `OCTOP_MEMORY_MCP_TOKEN=`（可写入 `~/.octop/env`，启动自动加载）。 +2. 重启 `octop run`。 +3. 验证：`curl -i http://:/mcp/memory/` 应返回 `401`（未带 token）或 MCP 协议响应（带 token）。 +4. 外部 agent 按 §1 连接信息接入。 + +--- + +## 5. 与进程内 MemoryService 的关系 + +| MCP 工具 | MemoryService 对应 | 层级 | +|---|---|---| +| `memory_capture` | `add_raw` | L0 原始事件 | +| `memory_extract` / `memory_promote` / `memory_reject` | 提取流水线（extract → candidate → promote） | L0 → L1 → L2 | +| `memory_raws` | 查询 raw events | L0 | +| `memory_candidates` | 查询候选队列 | L1 | +| `memory_recall` / `memory_save` / `memory_update` | 原子/树读写 | L2 | + +写入链路：`capture`（L0）→ `extract`（L1 候选）→ `promote`（L2 原子）；直接 `save` 则跳过提取直写 L2。 diff --git a/src/octop/api/app.py b/src/octop/api/app.py index d30f3599..5f5477f1 100644 --- a/src/octop/api/app.py +++ b/src/octop/api/app.py @@ -248,6 +248,23 @@ async def acme_http01_challenge(token: str) -> PlainTextResponse: ], ) + # 专家记忆 MCP server（对外暴露，独立 token 鉴权，未配置 OCTOP_MEMORY_MCP_TOKEN 时不挂载） + from octop.infra.agents.memory_mcp import mount_memory_mcp + + memory_mcp_managers = mount_memory_mcp(app, server) + if memory_mcp_managers: + from contextlib import AsyncExitStack, asynccontextmanager + + @asynccontextmanager + async def _memory_mcp_lifespan(application: FastAPI): + # streamable_http_app 的 task group 依赖 lifespan，挂载后须手动并入 + async with AsyncExitStack() as stack: + for mgr in memory_mcp_managers: + await stack.enter_async_context(mgr.run()) + yield + + app.router.lifespan_context = _memory_mcp_lifespan + if enable_api_docs: @app.get("/api/docs", include_in_schema=False) diff --git a/src/octop/infra/agents/memory_mcp.py b/src/octop/infra/agents/memory_mcp.py new file mode 100644 index 00000000..59b1697a --- /dev/null +++ b/src/octop/infra/agents/memory_mcp.py @@ -0,0 +1,657 @@ +"""Expose Octop expert memory as an MCP server for external agents. + +External agents (coding agents, bots) can read/write Octop expert memory over +MCP (Streamable HTTP), aligned with the in-process ``MemoryService`` +capabilities. Every write stamps a ``source`` marker that can be traced back +on recall. + +Expert binding: the endpoint is a single ``/mcp/memory`` mount; the expert is +selected at connect time via the ``X-Octop-Agent-Id`` header (one connection +binds one expert — the caller never passes an agent id per tool call). + +raw vs atom (aligned with ``MemoryService``): + +* ``memory_capture`` -> ``add_raw``: writes an **L0 raw event**, which goes + through the extraction pipeline (extract -> candidate -> promote -> atom). + Use it to record raw conversations / events. The record is visible + immediately via ``memory_raws``; ``memory_recall`` returns it only + after extraction promotes it to an atom. +* ``memory_save`` -> ``store``: persists a structured fact directly into the + canonical atom/tree (durable, no extraction). Use it when you already know + the exact fact to remember. + +Auth: independent token via ``OCTOP_MEMORY_MCP_TOKEN`` (fail-closed when +unset), enforced by the ASGI middleware in ``mount_memory_mcp``. +""" + +from __future__ import annotations + +import logging +import os +from contextvars import ContextVar +from typing import Any + +from mcp.server.fastmcp import FastMCP +from mcp.server.fastmcp.server import Context +from mcp.server.transport_security import TransportSecuritySettings + +from octop.infra.agents.memory_backend import open_memory_kwargs +from octop.infra.server import OctopServer + +logger = logging.getLogger(__name__) + +# 当前 MCP HTTP 请求的调用者 user id（由 _AgentRouter 中间件写入，工具读取）。 +# stateless streamable HTTP 下 mcp SDK 不提供 ctx.request_context，故用 contextvar +# 跨 ASGI 中间件 → 工具传递，供 memory_capture/save 做 per-user 追溯。 +_current_caller_user: ContextVar[str] = ContextVar("octop_mcp_caller_user", default="") + + +def _open_memory(server: OctopServer, agent_id: str) -> Any: + """Open the agent's ``Memory`` instance (sqlite by default, postgres opt-in). + + Mirrors ``api.common.memory_client._open_memory_for_agent`` but stays in + ``infra/`` (no api dependency). Workspace is resolved from the agent + registry, falling back to the Octop default layout. + """ + from harness_memory.core import Memory # noqa: PLC0415 + + services = server.services + assert services is not None, "server.services required for memory backend" + runtime = getattr(server, "app_runtime", None) + registry = getattr(runtime, "agent_registry", None) if runtime is not None else None + if registry is not None and hasattr(registry, "resolve_workspace_dir"): + workspace = registry.resolve_workspace_dir(agent_id) + else: + paths = getattr(server, "paths", None) or services.paths + workspace = paths.ensure_agent_workspace(agent_id) + + row = services.agent_repo.get(agent_id) + cfg: dict[str, Any] = {} + if row is not None and row.config_json: + import json # noqa: PLC0415 + + try: + parsed = json.loads(row.config_json) + if isinstance(parsed, dict): + cfg = parsed + except json.JSONDecodeError: + cfg = {} + + ns, backend, backend_config = open_memory_kwargs( + agent_id=agent_id, + cfg=cfg, + octop_config=services.config, + workspace_dir=workspace, + ) + return Memory(namespace=ns, backend=backend, backend_config=backend_config) + + +def build_memory_mcp(server: OctopServer, agent_id: str) -> FastMCP: + """Build an MCP server bound to one expert (``agent_id`` captured in closure).""" + mcp = FastMCP( + f"octop-memory-{agent_id}", + # Stateless streamable HTTP: every request gets a fresh transport, no + # Mcp-Session-Id tracking. Session state is in-memory per process, so a + # server restart silently orphans every client session id and the next + # tool call fails with -32600 "Session not found". Stateless mode + # eliminates that failure class entirely (clients re-initialize per + # request); the cost is one extra initialize per tool call. + stateless_http=True, + # Octop runs behind a reverse proxy (Host is the public domain, forwarded + # by nginx), not a localhost dev scenario — the mcp SDK's localhost + # DNS-rebinding protection does not apply and would reject the Host + # with 421 unless the domain is allow-listed. + transport_security=TransportSecuritySettings(enable_dns_rebinding_protection=False), + ) + # Collapse the streamable-HTTP path to "/" so the endpoint is exactly + # /mcp/memory (the default "/mcp" would make it /mcp/memory/mcp). + mcp.settings.streamable_http_path = "/" + + def _memory() -> Any: + return _open_memory(server, agent_id) + + def _caller_user(ctx: Any | None) -> str: + """读取当前 MCP 请求的调用者 user id。 + + 优先级：显式 ``user`` 参数 → ``X-Octop-User-Id`` header（由 + ``_AgentRouter`` 中间件写入 contextvar）。stateless HTTP 下 mcp SDK + 不提供 ``ctx.request_context``，故不依赖它。 + """ + try: + return _current_caller_user.get() or "" + except Exception: # noqa: BLE001 + return "" + + def _derive_session(source: str, user: str) -> str: + """外部调用缺省 session_id 时派生稳定会话键。 + + 规则 ``ext:{source}:{user}``：同 source 同 user 的多次 capture 落入 + 同一分组，harness 提取管线能聚合蒸馏成 atom；不同 source / 不同 user + 分开分组，避免混入彼此上下文。 + """ + return f"ext:{source or 'mcp'}:{user or 'anon'}" + + + @mcp.tool() + def memory_recall( + query: str, + limit: int = 5, + user: str | None = None, + ctx: Context | None = None, # type: ignore[type-arg] + ) -> dict[str, Any]: + """Recall memories from this expert (aligned with the in-process recall_inject). + + **日常使用**：每次对话/任务开始前调用，把专家记忆中与 query 相关的 + atom 召回注入上下文。运行完整召回管线（tokenize -> FTS -> rerank -> + dedupe），返回结构化片段 + 可注入 system prompt 的 markdown 块。 + + 调用者身份（``X-Octop-User-Id`` header 或 ``user`` 参数）会记录在 + 返回的 ``caller`` 字段，供按调用者追溯召回来源；记忆本身是专家级 + 共享，不按用户隔离。 + + Args: + query: free-form question / keywords (pass the whole sentence; the + pipeline tokenizes CJK into n-grams internally). + limit: max number of snippets to return. + user: optional caller id (overrides the ``X-Octop-User-Id`` header). + ctx: injected MCP context (reads ``X-Octop-User-Id`` header). + """ + from harness_memory.pipeline.recall import recall_for_prompt # noqa: PLC0415 + + caller = user or _caller_user(ctx) + memory = _memory() + result = recall_for_prompt(memory, query, limit=limit) + return { + "memories": [ + { + "source_id": s.source_id, + "timestamp": s.timestamp_iso, + "layer": s.layer, + "text": s.text, + } + for s in result.snippets + ], + "count": len(result.snippets), + "rendered": result.rendered, + "caller": caller or None, + } + + @mcp.tool() + def memory_save( + content: str, + source: str, + topic: str | None = None, + user: str | None = None, + ctx: Context | None = None, # type: ignore[type-arg] + ) -> dict[str, Any]: + """Persist a structured fact directly (atom/tree, durable, no extraction). + + **显式记忆**（非日常）：仅当你知道一个明确的、需要长期记住的事实 + 时才调用（如用户偏好、项目约定）。立即通过 ``memory_recall`` 可召回， + 不经过提取管线。日常对话内容请用 ``memory_capture`` 交给自动提取。 + The source marker is stored in ``metadata.source``; the caller id (from + ``X-Octop-User-Id`` header or ``user`` arg) is stored in ``metadata.user``. + + Args: + content: the fact to remember. + source: who/what recorded it (e.g. "coding-agent"), for traceability. + topic: optional topic label. + user: optional caller id (overrides the ``X-Octop-User-Id`` header). + ctx: injected MCP context (reads ``X-Octop-User-Id`` header). + """ + caller = user or _caller_user(ctx) + memory = _memory() + node = memory.store( + content, + topic=topic, + metadata={"source": source, **({"user": caller} if caller else {})}, + ) + return { + "node_id": node.id, + "content": node.content, + "source": source, + "user": caller or None, + } + + @mcp.tool() + def memory_capture( + content: str, + source: str, + session_id: str | None = None, + user: str | None = None, + ctx: Context | None = None, # type: ignore[type-arg] + ) -> dict[str, Any]: + """Record a raw event to L0 (goes through extraction: extract -> candidate -> atom). + + **日常使用**：把对话/事件原始内容记录下来，交给自动提取流水线 + （extract -> candidate -> promote -> atom），稍后经 ``memory_recall`` + 可召回。记录后立即可用 ``memory_raws`` 查询。The source marker + is stored in ``payload.source``; the caller id (from ``X-Octop-User-Id`` + header or ``user`` arg) is stored on the raw event for per-user + traceability. + + If ``session_id`` is omitted it is derived as ``ext:{source}:{user}`` + so external callers without an Octop native session still get their raw + events grouped and distilled into atoms (extraction groups by session). + + Example:: + + memory_capture( + content="user reported: the report panel banner is not rendering", + source="review-bot", + ) + # -> {"event_id": "...", "recorded": true, "extract_scheduled": true, ...} + # later: memory_recall(query="report panel banner not rendering") + + Args: + content: the raw conversation / event text. + source: who/what recorded it, for traceability. + session_id: optional stable session id (e.g. caller name) so the + extraction pipeline can group events by session. When omitted, + derived as ``ext:{source}:{user}``. + user: optional caller id (overrides the ``X-Octop-User-Id`` header). + ctx: injected MCP context (reads ``X-Octop-User-Id`` header). + """ + caller = user or _caller_user(ctx) + effective_session = session_id or _derive_session(source, caller) + memory = _memory() + raw = memory.add_raw( + content, + event_type="manual", + host="mcp-external", + session_id=effective_session, + user=caller or None, + payload={"source": source}, + ) + extract_scheduled = _trigger_extract(server, agent_id, effective_session) + return { + "event_id": raw.id, + "source": source, + "user": caller or None, + "session_id": effective_session, + "recorded": True, + "extract_scheduled": extract_scheduled, + "note": ( + "raw (L0) event recorded; visible now via memory_raws, " + "recallable via memory_recall after the extraction pipeline " + "promotes it to an atom" + ), + } + + @mcp.tool() + def memory_update( + atom_id: str, + new_content: str, + source: str, + note: str = "mcp update", + user: str | None = None, + ctx: Context | None = None, # type: ignore[type-arg] + ) -> dict[str, Any]: + """Update a memory: deprecate the old atom and persist the new fact. + + **显式更新**（非日常）：仅当已知旧记忆已过时、需要替换时才调用 + （如用户纠正了一个事实）。旧 atom 标记 deprecated，新事实立即经 + ``memory_recall`` 可召回。日常纠错也可以走 ``memory_capture`` 让 + 提取管线处理。 + + Args: + atom_id: id of the atom to supersede. + new_content: the replacement fact. + source: who/what updated it, for traceability. + note: deprecation note. + user: optional caller id (overrides the ``X-Octop-User-Id`` header). + ctx: injected MCP context (reads ``X-Octop-User-Id`` header). + """ + caller = user or _caller_user(ctx) + memory = _memory() + deprecated = memory.deprecate_atom(atom_id, actor="user", note=note) + node = memory.store( + new_content, + metadata={ + "source": source, + "supersedes": atom_id, + **({"user": caller} if caller else {}), + }, + ) + return { + "deprecated": deprecated, + "deprecated_atom_id": atom_id, + "new_node_id": node.id, + "source": source, + "user": caller or None, + } + + # ─── 记忆分层查询 + 流水线调度（L0/L1/L2）───────────────────────── + # 参考 DSH 记忆工具集（memory_raws/candidates/atoms/extract/promote/reject）， + # 把记忆生产流水线的每个环节暴露为 MCP 工具，供外部调度。 + + @mcp.tool() + def memory_raws( + query: str | None = None, + session_id: str | None = None, + host: str | None = None, + user: str | None = None, + limit: int = 50, + ctx: Context | None = None, # type: ignore[type-arg] + ) -> dict[str, Any]: + """**L0 原始事件查询**：FTS 搜索或结构化过滤原始事件（证据源）。 + + 合并了原 ``memory_search_raw``（全文搜索）与结构化过滤两种能力： + ``query`` 走 FTS 全文搜索（capture 立即可见，提取前也可查）， + ``session_id``/``host``/``user`` 做结构化过滤。按时间倒序返回。 + + Args: + query: FTS keywords to match raw event content (optional). + session_id: filter by session (e.g. ``ext:review-bot:user-alice``). + host: filter by recording host (e.g. ``mcp-external``). + user: filter by caller user id (also read from X-Octop-User-Id). + limit: max events (default 50). + ctx: injected MCP context. + """ + caller = user or _caller_user(ctx) + memory = _memory() + events = memory.search_raw(query, limit=limit) if query else memory.list_raw( + session_id=session_id, + host=host, + user=user or (caller or None), + limit=limit, + ) + return { + "events": [ + { + "event_id": e.id, + "timestamp": e.timestamp.isoformat(), + "session_id": e.session_id, + "user": e.user, + "event_type": e.event_type, + "source": (e.payload or {}).get("source") if e.payload else None, + "content": e.content, + } + for e in events + ], + "count": len(events), + "caller": caller or None, + } + + @mcp.tool() + def memory_candidates( + status: str | None = None, + session_id: str | None = None, + limit: int = 50, + ) -> dict[str, Any]: + """**L1 候选记忆列表**：查询待审核/已晋升/已拒绝的候选（提取产物）。 + + 候选由 ``memory_capture`` 触发的提取流水线生成（或手动 + ``memory_extract``）。默认返回 pending 队列；可用 ``status`` 过滤 + （pending / promoted / rejected / needs_review / conflict）。 + + Args: + status: filter by candidate status (default pending). + session_id: filter by source session. + limit: max candidates (default 50). + """ + memory = _memory() + from harness_memory.core import CandidateStatus # noqa: PLC0415 + + status_enum = None + if status: + try: + status_enum = CandidateStatus(status) + except ValueError: + status_enum = None + candidates = memory.list_candidates( + status=status_enum, + session_id=session_id, + limit=limit, + ) + return { + "candidates": [ + { + "candidate_id": c.id, + "status": c.status.value if hasattr(c.status, "value") else str(c.status), + "candidate_type": getattr(c, "candidate_type", None), + "assertion": getattr(c, "assertion", None), + "session_id": getattr(c, "session_id", None), + "confidence": getattr(c, "confidence", None), + } + for c in candidates + ], + "count": len(candidates), + } + + @mcp.tool() + def memory_extract( + session_id: str | None = None, + limit: int = 100, + promote: bool = False, + ) -> dict[str, Any]: + """**手动触发记忆提取**（L0 → L1，可选直达 L2）：调度生产流水线。 + + 复用专家进程内 ``MemoryService``（含配置的提取 LLM）同步执行提取： + 取最近 ``limit`` 条 L0 原始事件 → LLM 类型化提取 → 候选（pending）； + ``promote=True`` 时对候选执行晋升检查（L1 → L2 atom），跳过人工审核。 + 运行时无 MemoryService 时返回 ``error``（best-effort）。 + + Args: + session_id: only extract events of this session; omit for recent all. + limit: number of recent raw events to extract (default 100). + promote: run promotion on extracted candidates (default False). + """ + runtime_server = server.app_runtime + assert runtime_server is not None, "app_runtime required for memory extract" + agent = runtime_server.agent_registry.get_agent(agent_id) + runtime = getattr(agent, "_memory_runtime", None) + service = getattr(runtime, "service", None) if runtime else None + if service is None: + return {"error": "MemoryService unavailable (agent not running / no memory runtime)"} + + eff_session = session_id or "manual" + result = service.extract(eff_session, incremental=True, promote=promote, regen_pages=False) + if not isinstance(result, dict): + return {"session_id": eff_session, "candidates": 0, "promoted": 0} + return { + "session_id": eff_session, + "events_considered": result.get("events_considered", 0), + "candidates": result.get("candidates", 0), + "promoted": result.get("promoted", 0) if isinstance(result.get("promotion"), dict) else 0, + "error": result.get("failure_reason"), + } + + @mcp.tool() + def memory_promote( + candidate_ids: list[str], + importance: str | None = None, + ) -> dict[str, Any]: + """**审核晋升候选**（L1 → L2）：确认候选为原子记忆。 + + 对指定候选执行 5 项晋升检查（规则路径），通过则写入 L2 atom， + 记录 journal。用于人工审核 / 外部调度晋升。 + + Args: + candidate_ids: candidate ids to promote (from memory_candidates). + importance: override importance (low/medium/high); default keep. + """ + memory = _memory() + candidates = memory.list_candidates(limit=1000) + by_id = {c.id: c for c in candidates} + selected = [by_id[cid] for cid in candidate_ids if cid in by_id] + if not selected: + return {"promoted": 0, "skipped": len(candidate_ids)} + result = memory.promote_candidates(selected) + return { + "promoted": result.promoted if hasattr(result, "promoted") else len(selected), + "skipped": len(candidate_ids) - len(selected), + } + + @mcp.tool() + def memory_reject( + candidate_id: str, + reason: str = "rejected by external caller", + ) -> dict[str, Any]: + """**拒绝候选**：标记 rejected + 原因，不进原子层（记录 journal 可审计）。 + + Args: + candidate_id: candidate id to reject. + reason: rejection reason. + """ + memory = _memory() + from harness_memory.core import CandidateStatus # noqa: PLC0415 + + ok = memory.update_candidate_status( + candidate_id, + status=CandidateStatus.REJECTED, + decided_by="mcp-external", + promotion_reason=reason, + ) + return {"ok": ok, "candidate_id": candidate_id, "status": "rejected"} + + return mcp + + +def _trigger_extract(server: OctopServer, agent_id: str, session_id: str | None) -> bool: + """Best-effort: asynchronously trigger the agent's memory extraction. + + Internal-network enhancement (not part of the community PR): raw events + written by MCP capture are not in the harness-agent extractor's tracked + sessions, so they would never be distilled into atoms. Reuse the agent's + in-process ``MemoryService`` (with the agent's configured extraction LLM) + via ``agent._memory_runtime.service`` (no public entrypoint; best-effort). + Returns whether an extract task was scheduled. + """ + import asyncio + + if not session_id: + return False + try: + runtime_server = server.app_runtime + assert runtime_server is not None, "app_runtime required for memory extract" + agent = runtime_server.agent_registry.get_agent(agent_id) + runtime = getattr(agent, "_memory_runtime", None) + service = getattr(runtime, "service", None) if runtime else None + if service is None: + return False + + async def _extract() -> None: + try: + await asyncio.to_thread( + service.extract, + session_id, + incremental=True, + promote=True, + regen_pages=True, + ) + except Exception: + logger.warning( + "memory extract failed for session %s", session_id, exc_info=True + ) + + asyncio.create_task(_extract()) + return True + except Exception: + logger.debug("memory extract trigger skipped for agent %s", agent_id, exc_info=True) + return False + + +def _memory_mcp_token() -> str | None: + """Read the MCP auth token (empty string treated as unconfigured).""" + return (os.environ.get("OCTOP_MEMORY_MCP_TOKEN") or "").strip() or None + + +class _TokenAuthMiddleware: + """ASGI middleware enforcing ``Authorization: Bearer`` or ``X-Octop-Memory-Token``.""" + + def __init__(self, app: Any, token: str) -> None: + self._app = app + self._token = token + + async def __call__(self, scope: dict[str, Any], receive: Any, send: Any) -> None: + if scope.get("type") != "http": + await self._app(scope, receive, send) + return + + headers = {k.decode("latin-1").lower(): v.decode("latin-1") for k, v in scope.get("headers", [])} + auth = headers.get("authorization", "") + provided = auth[7:].strip() if auth.startswith("Bearer ") else "" + if not provided: + provided = headers.get("x-octop-memory-token", "").strip() + + if provided != self._token: + body = b'{"error":"unauthorized"}' + await send({ + "type": "http.response.start", + "status": 401, + "headers": [ + (b"content-type", b"application/json"), + (b"content-length", str(len(body)).encode()), + ], + }) + await send({"type": "http.response.body", "body": body}) + return + + await self._app(scope, receive, send) + + +class _AgentRouter: + """ASGI dispatcher routing to the per-expert MCP app by ``X-Octop-Agent-Id`` header.""" + + def __init__(self, mcp_apps: dict[str, Any]) -> None: + self._mcp_apps = mcp_apps + + async def __call__(self, scope: dict[str, Any], receive: Any, send: Any) -> None: + if scope.get("type") != "http": + return # lifespan is wired into the host FastAPI manually; http only here + + headers = {k.decode("latin-1").lower(): v.decode("latin-1") for k, v in scope.get("headers", [])} + agent_id = headers.get("x-octop-agent-id", "").strip() + target = self._mcp_apps.get(agent_id) + if target is None: + body = b'{"error":"missing or unknown agent_id (X-Octop-Agent-Id)"}' + await send({ + "type": "http.response.start", + "status": 404, + "headers": [ + (b"content-type", b"application/json"), + (b"content-length", str(len(body)).encode()), + ], + }) + await send({"type": "http.response.body", "body": body}) + return + # 把调用者 user id 写入 contextvar，供工具读取（stateless HTTP 下 + # mcp SDK 不提供 ctx.request_context）。 + user = headers.get("x-octop-user-id", "").strip() + token_cv = _current_caller_user.set(user) + try: + await target(scope, receive, send) + finally: + _current_caller_user.reset(token_cv) + + +def mount_memory_mcp(app: Any, server: OctopServer) -> list[Any]: + """Mount the memory MCP endpoint at ``/mcp/memory``; the expert is selected + per connection via the ``X-Octop-Agent-Id`` header (one connection binds one + expert; the URL stays uniform and does not leak expert ids). + + Does not mount when ``OCTOP_MEMORY_MCP_TOKEN`` is unset (fail-closed). + Returns the session managers that must be initialized in the host FastAPI + lifespan (``streamable_http_app`` task groups depend on it). + """ + token = _memory_mcp_token() + if token is None: + return [] + + managers: list[Any] = [] + mcp_apps: dict[str, Any] = {} + services = server.services + assert services is not None, "server.services required for memory MCP mount" + rows = services.agent_repo.list_all(include_disabled=False) + for row in rows: + agent_id = row.agent_id + mcp = build_memory_mcp(server, agent_id) + mcp_apps[agent_id] = mcp.streamable_http_app() + managers.append(mcp._session_manager) + + app.mount("/mcp/memory", _TokenAuthMiddleware(_AgentRouter(mcp_apps), token)) + return managers + + +__all__ = ["build_memory_mcp", "mount_memory_mcp"] diff --git a/tests/unit/agents/test_memory_mcp.py b/tests/unit/agents/test_memory_mcp.py new file mode 100644 index 00000000..9985dc30 --- /dev/null +++ b/tests/unit/agents/test_memory_mcp.py @@ -0,0 +1,274 @@ +"""Unit tests for the expert memory MCP server (infra/agents/memory_mcp).""" + +from __future__ import annotations + +from unittest import mock + +import pytest + +from octop.infra.agents import memory_mcp as mm + + +@pytest.fixture +def fake_memory(monkeypatch): + mem = mock.MagicMock() + mem.recall.return_value = [] + node = mock.MagicMock() + node.id = "node1" + node.content = "remember X" + mem.store.return_value = node + mem.add_raw.return_value = mock.MagicMock(id="evt1") + mem.deprecate_atom.return_value = True + monkeypatch.setattr(mm, "_open_memory", lambda server, agent_id: mem) + return mem + + +def _tools(mcp): + return mcp._tool_manager._tools + + +def test_build_binds_agent_id(monkeypatch): + """Tools capture agent_id in the closure; callers never pass it.""" + captured = {} + + def fake_open(server, agent_id): + captured["agent_id"] = agent_id + mem = mock.MagicMock() + mem.store.return_value = mock.MagicMock(id="n1", content="x") + return mem + + monkeypatch.setattr(mm, "_open_memory", fake_open) + mcp = mm.build_memory_mcp(mock.MagicMock(), agent_id="EXPERT42") + _tools(mcp)["memory_save"].fn(content="x", source="s") + assert captured["agent_id"] == "EXPERT42" + + +def test_build_registers_five_tools(fake_memory): + mcp = mm.build_memory_mcp(mock.MagicMock(), "A1") + assert set(_tools(mcp)) == { + "memory_recall", + "memory_save", + "memory_capture", + "memory_update", + "memory_raws", + "memory_candidates", + "memory_extract", + "memory_promote", + "memory_reject", + } + + +def test_memory_recall_uses_full_pipeline(fake_memory, monkeypatch): + """memory_recall runs the full recall pipeline (recall_for_prompt).""" + import harness_memory.pipeline.recall as _recall + + class _Snippet: + source_id = "atom-1" + timestamp_iso = "2026-08-19T00:00:00+00:00" + layer = "atom" + text = "billing-migration is the local clone" + + fake_result = mock.MagicMock() + fake_result.snippets = [_Snippet()] + fake_result.rendered = "markdown" + monkeypatch.setattr(_recall, "recall_for_prompt", lambda m, q, limit: fake_result) + + mcp = mm.build_memory_mcp(mock.MagicMock(), "A1") + result = _tools(mcp)["memory_recall"].fn(query="billing-migration", limit=3) + assert result["count"] == 1 + assert result["memories"][0]["text"] == "billing-migration is the local clone" + assert result["rendered"] == "markdown" + + +def test_memory_save_goes_store(fake_memory): + mcp = mm.build_memory_mcp(mock.MagicMock(), "A1") + result = _tools(mcp)["memory_save"].fn(content="remember X", source="coding-agent") + kwargs = fake_memory.store.call_args.kwargs + assert kwargs["topic"] is None + assert kwargs["metadata"] == {"source": "coding-agent"} + assert result["source"] == "coding-agent" + + +def test_memory_capture_goes_add_raw(fake_memory): + mcp = mm.build_memory_mcp(mock.MagicMock(), "A1") + result = _tools(mcp)["memory_capture"].fn( + content="raw conversation", source="review-bot", session_id="review-1" + ) + kwargs = fake_memory.add_raw.call_args.kwargs + assert kwargs["event_type"] == "manual" + assert kwargs["host"] == "mcp-external" + assert kwargs["session_id"] == "review-1" + assert kwargs["payload"] == {"source": "review-bot"} + assert result["recorded"] is True + assert "raw (L0)" in result["note"] + + +def test_memory_raws_queries_l0_with_query(fake_memory): + class _Evt: + id = "evt1" + timestamp = __import__("datetime").datetime(2026, 8, 19) + session_id = "review-1" + user = "u1" + event_type = "manual" + payload = {"source": "review-bot"} + content = "report panel banner hidden" + + fake_memory.search_raw.return_value = [_Evt()] + mcp = mm.build_memory_mcp(mock.MagicMock(), "A1") + result = _tools(mcp)["memory_raws"].fn(query="report panel banner", limit=5) + fake_memory.search_raw.assert_called_once_with("report panel banner", limit=5) + assert result["count"] == 1 + assert result["events"][0]["event_id"] == "evt1" + assert result["events"][0]["source"] == "review-bot" + + +def test_memory_update_deprecates_and_saves(fake_memory): + mcp = mm.build_memory_mcp(mock.MagicMock(), "A1") + result = _tools(mcp)["memory_update"].fn( + atom_id="atom1", new_content="new fact", source="review-bot" + ) + fake_memory.deprecate_atom.assert_called_once_with("atom1", actor="user", note="mcp update") + assert fake_memory.store.call_args.kwargs["metadata"] == { + "source": "review-bot", + "supersedes": "atom1", + } + assert result["deprecated"] is True + + +def _asgi_scope(headers: list[tuple[bytes, bytes]] | None = None) -> dict: + return {"type": "http", "headers": headers or []} + + +@pytest.mark.asyncio +async def test_token_middleware_rejects_bad_token(): + inner_called = False + + async def _inner(scope, receive, send): + nonlocal inner_called + inner_called = True + + mw = mm._TokenAuthMiddleware(_inner, "secret") + sent = [] + scope = _asgi_scope([(b"authorization", b"Bearer wrong")]) + + async def _send(msg): + sent.append(msg) + + await mw(scope, lambda: {}, _send) + assert inner_called is False + assert sent[0]["status"] == 401 + + +@pytest.mark.asyncio +async def test_token_middleware_accepts_bearer(): + inner_called = False + + async def _inner(scope, receive, send): + nonlocal inner_called + inner_called = True + + mw = mm._TokenAuthMiddleware(_inner, "secret") + scope = _asgi_scope([(b"authorization", b"Bearer secret")]) + await mw(scope, lambda: {}, lambda msg: None) + assert inner_called is True + + +def test_mount_fail_closed_without_token(monkeypatch): + monkeypatch.delenv("OCTOP_MEMORY_MCP_TOKEN", raising=False) + app = mock.MagicMock() + assert mm.mount_memory_mcp(app, mock.MagicMock()) == [] + app.mount.assert_not_called() + + +def test_mount_unified_path_with_header_router(monkeypatch): + from types import SimpleNamespace + + monkeypatch.setenv("OCTOP_MEMORY_MCP_TOKEN", "secret") + app = mock.MagicMock() + server = SimpleNamespace( + services=SimpleNamespace( + agent_repo=mock.MagicMock( + list_all=lambda include_disabled: [ + SimpleNamespace(agent_id="A1"), + SimpleNamespace(agent_id="A2"), + ] + ) + ) + ) + managers = mm.mount_memory_mcp(app, server) + assert len(managers) == 2 + # unified path mounted exactly once + app.mount.assert_called_once() + assert app.mount.call_args.args[0] == "/mcp/memory" + + +@pytest.mark.asyncio +async def test_agent_router_routes_by_header(): + """_AgentRouter routes to the right app by X-Octop-Agent-Id header.""" + called = {} + + class _FakeApp: + def __init__(self, aid): + self._aid = aid + + async def __call__(self, scope, receive, send): + called["agent"] = self._aid + + router = mm._AgentRouter({"A1": _FakeApp("A1"), "A2": _FakeApp("A2")}) + scope = {"type": "http", "headers": [(b"x-octop-agent-id", b"A2")]} + await router(scope, lambda: {}, lambda msg: None) + assert called["agent"] == "A2" + + +@pytest.mark.asyncio +async def test_agent_router_404_unknown_agent(): + """Unknown agent_id returns 404.""" + router = mm._AgentRouter({"A1": mock.MagicMock()}) + scope = {"type": "http", "headers": [(b"x-octop-agent-id", b"NOPE")]} + sent = [] + + async def _send(msg): + sent.append(msg) + + await router(scope, lambda: {}, _send) + assert sent[0]["status"] == 404 + + +def test_trigger_extract_no_session_returns_false(): + """No session_id -> no extraction trigger.""" + assert mm._trigger_extract(mock.MagicMock(), "A1", None) is False + + +def test_trigger_extract_no_service_returns_false(monkeypatch): + """Agent without memory runtime/service -> silently skipped.""" + agent = mock.MagicMock() + runtime = mock.MagicMock() + runtime.service = None + agent._memory_runtime = runtime + registry = mock.MagicMock(get_agent=lambda aid: agent) + server = mock.MagicMock() + server.app_runtime.agent_registry = registry + assert mm._trigger_extract(server, "A1", "kiro-chat") is False + + +def test_trigger_extract_schedules_service(monkeypatch): + """With a service, asynchronously schedule extract and return True.""" + import asyncio + import time + + agent = mock.MagicMock() + service = mock.MagicMock() + runtime = mock.MagicMock() + runtime.service = service + agent._memory_runtime = runtime + registry = mock.MagicMock(get_agent=lambda aid: agent) + server = mock.MagicMock() + server.app_runtime.agent_registry = registry + + async def _run(): + return mm._trigger_extract(server, "A1", "kiro-chat") + + assert asyncio.run(_run()) is True + time.sleep(0.1) + service.extract.assert_called() + assert service.extract.call_args.args[0] == "kiro-chat"