From ba56646277508ebcd5b34780e7c786e3d9909b34 Mon Sep 17 00:00:00 2001 From: "Arvin.qi" Date: Thu, 20 Aug 2026 13:20:55 +0800 Subject: [PATCH 1/8] feat(memory): expose expert memory as an MCP server for external agents Add a memory MCP server (Streamable HTTP at /mcp/memory) so external agents (coding agents, bots) can read/write Octop expert memory directly, aligned with the in-process MemoryService capabilities. Tools (per expert, bound at connect time via X-Octop-Agent-Id header): - memory_recall(query, limit): full recall pipeline (tokenize + FTS + rerank), returns structured memories + rendered markdown - memory_save(content, source, topic?): persist a structured fact directly into the atom/tree (durable, no extraction) - memory_capture(content, source, session_id?): write an L0 raw event (extraction pipeline); visible immediately via memory_search_raw - memory_search_raw(query, limit): FTS-search L0 raw events (capture visible before extraction) - memory_update(atom_id, new_content, source): deprecate old atom + save new Auth: independent token via OCTOP_MEMORY_MCP_TOKEN (fail-closed if unset); authorization via Authorization: Bearer or X-Octop-Memory-Token. Implementation: - Lives in infra/agents/memory_mcp.py (no api-layer dependency; opens the agent Memory instance via open_memory_kwargs) - One FastMCP per agent, routed by X-Octop-Agent-Id header at /mcp/memory - DNS rebinding protection disabled (server runs behind a reverse proxy) - streamable_http task groups wired into the FastAPI lifespan Tests: tests/unit/agents/test_memory_mcp.py (tools, header routing, token middleware, unified mount). --- src/octop/api/app.py | 17 ++ src/octop/infra/agents/memory_mcp.py | 344 +++++++++++++++++++++++++++ tests/unit/agents/test_memory_mcp.py | 229 ++++++++++++++++++ 3 files changed, 590 insertions(+) create mode 100644 src/octop/infra/agents/memory_mcp.py create mode 100644 tests/unit/agents/test_memory_mcp.py diff --git a/src/octop/api/app.py b/src/octop/api/app.py index 14ba5e8f..486e4c35 100644 --- a/src/octop/api/app.py +++ b/src/octop/api/app.py @@ -233,6 +233,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..2a4a94cc --- /dev/null +++ b/src/octop/infra/agents/memory_mcp.py @@ -0,0 +1,344 @@ +"""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_search_raw``; ``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 typing import Any + +from mcp.server.fastmcp import FastMCP +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__) + + +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 + + 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 server.services.paths + workspace = paths.ensure_agent_workspace(agent_id) + + row = server.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=server.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}", + # 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(): + return _open_memory(server, agent_id) + + @mcp.tool() + def memory_recall(query: str, limit: int = 5) -> dict[str, Any]: + """Recall memories from this expert (aligned with the in-process recall_inject). + + Runs the full recall pipeline (tokenization -> FTS -> rerank -> dedupe) + and returns structured snippets plus a rendered markdown block ready to + inject into a system prompt. + + 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. + """ + from harness_memory.pipeline.recall import recall_for_prompt # noqa: PLC0415 + + 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, + } + + @mcp.tool() + def memory_save( + content: str, + source: str, + topic: str | None = None, + ) -> dict[str, Any]: + """Persist a structured fact directly (atom/tree, durable, no extraction). + + Use this when you already know the exact fact to remember — it is + immediately recallable via ``memory_recall``. The source marker is + stored in ``metadata.source``. + + Args: + content: the fact to remember. + source: who/what recorded it (e.g. "coding-agent"), for traceability. + topic: optional topic label. + """ + memory = _memory() + node = memory.store(content, topic=topic, metadata={"source": source}) + return {"node_id": node.id, "content": node.content, "source": source} + + @mcp.tool() + def memory_capture(content: str, source: str, session_id: str | None = None) -> dict[str, Any]: + """Record a raw event to L0 (goes through extraction: extract -> candidate -> atom). + + Use this to record raw conversations / events that the extraction + pipeline will later distill into atoms. The record is NOT immediately + recallable via ``memory_recall`` (that reads atoms); query it right + away with ``memory_search_raw``. The source marker is stored in + ``payload.source``. + + Example:: + + memory_capture( + content="user reported: the report panel banner is not rendering", + source="review-bot", + session_id="review-2026-08-20", + ) + # -> {"event_id": "...", "recorded": 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. + """ + memory = _memory() + raw = memory.add_raw( + content, + event_type="manual", + host="mcp-external", + session_id=session_id, + payload={"source": source}, + ) + return { + "event_id": raw.id, + "source": source, + "recorded": True, + "note": ( + "raw (L0) event recorded; visible now via memory_search_raw, " + "recallable via memory_recall after the extraction pipeline " + "promotes it to an atom" + ), + } + + @mcp.tool() + def memory_search_raw(query: str, limit: int = 10) -> dict[str, Any]: + """FTS-search L0 raw events of this expert (capture visible immediately). + + Unlike ``memory_recall`` (which reads atoms), this searches the raw + event layer, so records written by ``memory_capture`` are visible right + away, before extraction promotes them. + + Args: + query: keywords to match against raw event content. + limit: max number of events to return. + """ + memory = _memory() + events = memory.search_raw(query, limit=limit) + return { + "events": [ + { + "event_id": e.id, + "timestamp": e.timestamp.isoformat(), + "session_id": e.session_id, + "user": e.user, + "source": (e.payload or {}).get("source") if e.payload else None, + "content": e.content, + } + for e in events + ], + "count": len(events), + } + + @mcp.tool() + def memory_update( + atom_id: str, + new_content: str, + source: str, + note: str = "mcp update", + ) -> dict[str, Any]: + """Update a memory: deprecate the old atom and persist the new fact. + + Args: + atom_id: id of the atom to supersede. + new_content: the replacement fact. + source: who/what updated it, for traceability. + note: deprecation note. + """ + memory = _memory() + deprecated = memory.deprecate_atom(atom_id, actor="user", note=note) + node = memory.store(new_content, metadata={"source": source, "supersedes": atom_id}) + return { + "deprecated": deprecated, + "deprecated_atom_id": atom_id, + "new_node_id": node.id, + "source": source, + } + + return mcp + + +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 + await target(scope, receive, send) + + +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] = {} + rows = server.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..1be2924c --- /dev/null +++ b/tests/unit/agents/test_memory_mcp.py @@ -0,0 +1,229 @@ +"""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_search_raw", + } + + +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_search_raw_queries_l0(fake_memory): + class _Evt: + id = "evt1" + timestamp = __import__("datetime").datetime(2026, 8, 19) + session_id = "review-1" + user = "u1" + 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_search_raw"].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 From 004714a30dfa538c2026a63a7eb2fe1766524910 Mon Sep 17 00:00:00 2001 From: jinlongqi Date: Sun, 23 Aug 2026 11:34:43 +0800 Subject: [PATCH 2/8] style: format memory_mcp.py with ruff --- src/octop/infra/agents/memory_mcp.py | 44 ++++++++++++++++------------ 1 file changed, 26 insertions(+), 18 deletions(-) diff --git a/src/octop/infra/agents/memory_mcp.py b/src/octop/infra/agents/memory_mcp.py index 2a4a94cc..0cd9152d 100644 --- a/src/octop/infra/agents/memory_mcp.py +++ b/src/octop/infra/agents/memory_mcp.py @@ -265,7 +265,9 @@ async def __call__(self, scope: dict[str, Any], receive: Any, send: Any) -> None await self._app(scope, receive, send) return - headers = {k.decode("latin-1").lower(): v.decode("latin-1") for k, v in scope.get("headers", [])} + 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: @@ -273,14 +275,16 @@ async def __call__(self, scope: dict[str, Any], receive: Any, send: Any) -> None 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.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 @@ -297,19 +301,23 @@ 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", [])} + 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.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 await target(scope, receive, send) From fea4c5976d79bdd8696758a8abb99f1f063c3c4e Mon Sep 17 00:00:00 2001 From: "Arvin.qi" Date: Sun, 23 Aug 2026 12:56:29 +0800 Subject: [PATCH 3/8] fix: resolve mypy type errors in memory_mcp.py - Add assert for server.services to satisfy mypy strict mode - Add return type annotation to _memory closure --- PR.md | 113 +++++++++++++++++++++++++++ src/octop/infra/agents/memory_mcp.py | 17 ++-- 2 files changed, 125 insertions(+), 5 deletions(-) create mode 100644 PR.md diff --git a/PR.md b/PR.md new file mode 100644 index 00000000..d843eb44 --- /dev/null +++ b/PR.md @@ -0,0 +1,113 @@ +# PR Title + +feat(memory): expose expert memory as an MCP server for external agents + +--- + +## Summary + +Adds a memory **MCP server** (Streamable HTTP at `/mcp/memory`) so external +agents (coding agents, bots, other AI tools) can directly **read / write / +update Octop expert memory**, aligned 1:1 with the in-process +`MemoryService` capabilities. Every write stamps a `source` marker that is +traceable on recall. + +## Why + +Octop experts accumulate rich memory (facts, conversations, decisions), but +today only the Octop dashboard / in-process agent can access it. External +agents that need to reuse that expertise (e.g. a coding agent asking a +business expert's accumulated knowledge) have no way in. This PR exposes the +same memory surface over the standard MCP protocol so any MCP-capable agent +can join the loop. + +## What + +- **New module** `src/octop/infra/agents/memory_mcp.py` — FastMCP server + bound to one expert per connection, plus token auth and header routing. +- **Mount** in `api/app.py` (`build_app`) at `/mcp/memory`, with + `streamable_http` task groups wired into the FastAPI lifespan. +- **Tests** `tests/unit/agents/test_memory_mcp.py` (13 tests). + +### Tools + +| Tool | Purpose | Backing API | +|------|---------|-------------| +| `memory_recall(query, limit=5)` | Recall memories (full pipeline: tokenize → FTS → rerank → dedupe); returns structured snippets + rendered markdown | `recall_for_prompt` | +| `memory_save(content, source, topic?)` | Persist a structured fact directly into the atom/tree (durable, no extraction) | `Memory.store` | +| `memory_capture(content, source, session_id?)` | Write an **L0 raw event** (goes through extraction); visible immediately via `memory_search_raw` | `Memory.add_raw` | +| `memory_search_raw(query, limit=10)` | FTS-search L0 raw events (capture visible before extraction) | `Memory.search_raw` | +| `memory_update(atom_id, new_content, source)` | Deprecate old atom + persist the new fact | `deprecate_atom` + `store` | + +### Expert binding & auth + +- **One connection binds one expert**: endpoint is a single `/mcp/memory`; + the expert is selected at connect time via the `X-Octop-Agent-Id` header — + callers never pass an agent id per tool call (they don't know the id list). +- **Auth**: independent token via `OCTOP_MEMORY_MCP_TOKEN` (fail-closed when + unset). Authorization via `Authorization: Bearer` or `X-Octop-Memory-Token`. + +### raw vs atom (for callers) + +- `memory_capture` → **L0 raw event** (evidence layer), distilled later by + the extraction pipeline (`extract → candidate → promote → atom`). Use it to + record raw conversations/events; the record is visible immediately via + `memory_search_raw` and recallable via `memory_recall` once promoted. +- `memory_save` → **atom/tree directly** (durable, no extraction). Use it + when the fact is already known. + +## Implementation notes + +- Lives in `infra/agents/` with no api-layer dependency: opens the agent + `Memory` instance via `open_memory_kwargs` + `Memory(...)` (workspace + resolved from the agent registry). +- DNS rebinding protection disabled (`TransportSecuritySettings`) because + Octop runs behind a reverse proxy (Host is the public domain, not localhost). +- `streamable_http_path` collapsed to `/` so the endpoint is exactly + `/mcp/memory` (the SDK default `/mcp` would yield `/mcp/memory/mcp`). +- One `FastMCP` per expert, routed by an ASGI dispatcher on the + `X-Octop-Agent-Id` header; missing/unknown agent → 404. + +## Usage example + +```bash +export OCTOP_MEMORY_MCP_TOKEN="" +``` + +```json +{ + "mcpServers": { + "octop-memory": { + "type": "streamable_http", + "url": "http:///mcp/memory/", + "headers": { + "Authorization": "Bearer ", + "X-Octop-Agent-Id": "" + } + } + } +} +``` + +```text +memory_recall(query="what are the key project decisions?") +memory_save(content="the release window is every Tuesday", source="coding-agent", topic="release") +memory_capture(content="user reported: the report panel banner is not rendering", source="review-bot", session_id="review-2026-08-20") +memory_search_raw(query="report panel banner") +memory_update(atom_id="atom_xxx", new_content="updated fact", source="coding-agent") +``` + +## Testing + +- `tests/unit/agents/test_memory_mcp.py` — 13 tests: tool registration, + recall pipeline, capture (raw) semantics, search_raw, update, token + middleware (401 / accept), header routing, 404 unknown agent, unified mount. +- Verified locally by booting the server and exercising the MCP endpoints: + health, 401 without token, `initialize` (binds expert via header), + `tools/list` (5 tools), `tools/call memory_recall`. + +## Checklist + +- [x] No internal/hard-coded environment-specific values in the diff +- [x] `make lint` clean (ruff) +- [x] Unit tests pass diff --git a/src/octop/infra/agents/memory_mcp.py b/src/octop/infra/agents/memory_mcp.py index 0cd9152d..931208ec 100644 --- a/src/octop/infra/agents/memory_mcp.py +++ b/src/octop/infra/agents/memory_mcp.py @@ -53,10 +53,14 @@ def _open_memory(server: OctopServer, agent_id: str) -> Any: 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 server.services.paths + services = server.services + assert services is not None, "server.services required when agent_registry unavailable" + paths = getattr(server, "paths", None) or services.paths workspace = paths.ensure_agent_workspace(agent_id) - row = server.services.agent_repo.get(agent_id) + services = server.services + assert services is not None, "server.services required for memory backend" + row = services.agent_repo.get(agent_id) cfg: dict[str, Any] = {} if row is not None and row.config_json: import json # noqa: PLC0415 @@ -71,7 +75,7 @@ def _open_memory(server: OctopServer, agent_id: str) -> Any: ns, backend, backend_config = open_memory_kwargs( agent_id=agent_id, cfg=cfg, - octop_config=server.services.config, + octop_config=services.config, workspace_dir=workspace, ) return Memory(namespace=ns, backend=backend, backend_config=backend_config) @@ -91,7 +95,7 @@ def build_memory_mcp(server: OctopServer, agent_id: str) -> FastMCP: # /mcp/memory (the default "/mcp" would make it /mcp/memory/mcp). mcp.settings.streamable_http_path = "/" - def _memory(): + def _memory() -> Any: return _open_memory(server, agent_id) @mcp.tool() @@ -336,9 +340,12 @@ def mount_memory_mcp(app: Any, server: OctopServer) -> list[Any]: if token is None: return [] + services = server.services + assert services is not None, "server.services required for memory MCP mount" + managers: list[Any] = [] mcp_apps: dict[str, Any] = {} - rows = server.services.agent_repo.list_all(include_disabled=False) + 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) From 0e7eec8c0054891ed033905047af704846c9ffb6 Mon Sep 17 00:00:00 2001 From: "Arvin.qi" Date: Wed, 26 Aug 2026 13:48:59 +0800 Subject: [PATCH 4/8] =?UTF-8?q?feat(memory-mcp):=20=E5=86=85=E7=BD=91?= =?UTF-8?q?=E5=A2=9E=E5=BC=BA=E2=80=94=E2=80=94=E8=B0=83=E7=94=A8=E8=80=85?= =?UTF-8?q?=20user=20=E8=BF=BD=E6=BA=AF=20+=20=E8=87=AA=E5=8A=A8=E6=8F=90?= =?UTF-8?q?=E5=8F=96=20+=20=E5=B7=A5=E5=85=B7=E6=8F=8F=E8=BF=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 基于社区 feat/memory-mcp-server (fea4c59) 的独立扩展分支，仅改 memory_mcp.py： - memory_capture 写 L0 后自动触发提取流水线（extract -> promote -> atom） - 缺省 session 派生 ext:{source}:{user}，外部调用无需 octop 原生 session - X-Octop-User-Id header / user 参数，全部工具支持调用者追溯 - 工具描述区分日常（recall/capture）与显式（save/update） - stateless HTTP + mypy strict 修复 --- src/octop/infra/agents/memory_mcp.py | 263 +++++++++++++++++++++------ 1 file changed, 208 insertions(+), 55 deletions(-) diff --git a/src/octop/infra/agents/memory_mcp.py b/src/octop/infra/agents/memory_mcp.py index 931208ec..fb56a508 100644 --- a/src/octop/infra/agents/memory_mcp.py +++ b/src/octop/infra/agents/memory_mcp.py @@ -28,9 +28,11 @@ 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 @@ -38,6 +40,11 @@ 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). @@ -48,18 +55,16 @@ def _open_memory(server: OctopServer, agent_id: str) -> Any: """ 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: - services = server.services - assert services is not None, "server.services required when agent_registry unavailable" paths = getattr(server, "paths", None) or services.paths workspace = paths.ensure_agent_workspace(agent_id) - services = server.services - assert services is not None, "server.services required for memory backend" row = services.agent_repo.get(agent_id) cfg: dict[str, Any] = {} if row is not None and row.config_json: @@ -85,6 +90,13 @@ 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 @@ -98,21 +110,55 @@ def build_memory_mcp(server: OctopServer, agent_id: str) -> FastMCP: 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) -> dict[str, Any]: + 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). - Runs the full recall pipeline (tokenization -> FTS -> rerank -> dedupe) - and returns structured snippets plus a rendered markdown block ready to - inject into a system prompt. + **日常使用**：每次对话/任务开始前调用，把专家记忆中与 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 { @@ -127,6 +173,7 @@ def memory_recall(query: str, limit: int = 5) -> dict[str, Any]: ], "count": len(result.snippets), "rendered": result.rendered, + "caller": caller or None, } @mcp.tool() @@ -134,60 +181,96 @@ 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). - Use this when you already know the exact fact to remember — it is - immediately recallable via ``memory_recall``. The source marker is - stored in ``metadata.source``. + **显式记忆**（非日常）：仅当你知道一个明确的、需要长期记住的事实 + 时才调用（如用户偏好、项目约定）。立即通过 ``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}) - return {"node_id": node.id, "content": node.content, "source": source} + 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) -> dict[str, Any]: + 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). - Use this to record raw conversations / events that the extraction - pipeline will later distill into atoms. The record is NOT immediately - recallable via ``memory_recall`` (that reads atoms); query it right - away with ``memory_search_raw``. The source marker is stored in - ``payload.source``. + **日常使用**：把对话/事件原始内容记录下来，交给自动提取流水线 + （extract -> candidate -> promote -> atom），稍后经 ``memory_recall`` + 可召回。记录后立即可用 ``memory_search_raw`` 查询。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", - session_id="review-2026-08-20", ) - # -> {"event_id": "...", "recorded": true, ...} + # -> {"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. + 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=session_id, + 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_search_raw, " "recallable via memory_recall after the extraction pipeline " @@ -196,7 +279,12 @@ def memory_capture(content: str, source: str, session_id: str | None = None) -> } @mcp.tool() - def memory_search_raw(query: str, limit: int = 10) -> dict[str, Any]: + def memory_search_raw( + query: str, + limit: int = 10, + user: str | None = None, + ctx: Context | None = None, # type: ignore[type-arg] + ) -> dict[str, Any]: """FTS-search L0 raw events of this expert (capture visible immediately). Unlike ``memory_recall`` (which reads atoms), this searches the raw @@ -206,7 +294,11 @@ def memory_search_raw(query: str, limit: int = 10) -> dict[str, Any]: Args: query: keywords to match against raw event content. limit: max number of events to return. + user: optional caller id (overrides the ``X-Octop-User-Id`` header); + returned in ``caller`` for per-caller traceability. + ctx: injected MCP context (reads ``X-Octop-User-Id`` header). """ + caller = user or _caller_user(ctx) memory = _memory() events = memory.search_raw(query, limit=limit) return { @@ -222,6 +314,7 @@ def memory_search_raw(query: str, limit: int = 10) -> dict[str, Any]: for e in events ], "count": len(events), + "caller": caller or None, } @mcp.tool() @@ -230,28 +323,90 @@ def memory_update( 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}) + 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, } 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 @@ -269,9 +424,7 @@ async def __call__(self, scope: dict[str, Any], receive: Any, send: Any) -> None await self._app(scope, receive, send) return - headers = { - k.decode("latin-1").lower(): v.decode("latin-1") for k, v in scope.get("headers", []) - } + 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: @@ -279,16 +432,14 @@ async def __call__(self, scope: dict[str, Any], receive: Any, send: Any) -> None 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.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 @@ -305,26 +456,29 @@ 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", []) - } + 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.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 - await target(scope, receive, send) + # 把调用者 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]: @@ -340,11 +494,10 @@ def mount_memory_mcp(app: Any, server: OctopServer) -> list[Any]: if token is None: return [] - services = server.services - assert services is not None, "server.services required for memory MCP mount" - 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 From 9160cbed14a459988743d5fd025eb9c21bd01433 Mon Sep 17 00:00:00 2001 From: "Arvin.qi" Date: Wed, 26 Aug 2026 15:12:04 +0800 Subject: [PATCH 5/8] =?UTF-8?q?feat(memory-mcp):=20=E8=AE=B0=E5=BF=86?= =?UTF-8?q?=E5=88=86=E5=B1=82=E5=B7=A5=E5=85=B7=E9=9B=86=20+=20=E7=94=9F?= =?UTF-8?q?=E4=BA=A7=E6=B5=81=E6=B0=B4=E7=BA=BF=E8=B0=83=E5=BA=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 参考 DSH 记忆工具设计，把记忆生产流水线每环节暴露为 MCP 工具： - memory_raws: L0 原始事件结构化列表（session/host/user 过滤） - memory_candidates: L1 候选队列查询（pending/promoted/rejected） - memory_extract: 手动调度提取（L0→L1，promote 可直达 L2） - memory_promote: 审核晋升候选（L1→L2） - memory_reject: 拒绝候选（标记+审计） - memory_atoms: L2 原子记忆结构化查询 全部工具经 MemoryService / Memory 分层 API 实现，与 capture 自动流水线互补， 支持外部按需调度记忆生产。 --- src/octop/infra/agents/memory_mcp.py | 218 +++++++++++++++++++++++++++ tests/unit/agents/test_memory_mcp.py | 46 ++++++ 2 files changed, 264 insertions(+) diff --git a/src/octop/infra/agents/memory_mcp.py b/src/octop/infra/agents/memory_mcp.py index fb56a508..2416e3d7 100644 --- a/src/octop/infra/agents/memory_mcp.py +++ b/src/octop/infra/agents/memory_mcp.py @@ -360,6 +360,224 @@ def memory_update( "user": caller or None, } + # ─── 记忆分层查询 + 流水线调度（L0/L1/L2）───────────────────────── + # 参考 DSH 记忆工具集（memory_raws/candidates/atoms/extract/promote/reject）， + # 把记忆生产流水线的每个环节暴露为 MCP 工具，供外部调度。 + + @mcp.tool() + def memory_raws( + 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 原始事件列表**：按 session/host/user 过滤查询原始事件（证据源）。 + + 与 ``memory_search_raw``（全文搜索）互补——本工具做结构化过滤 + （session/host/user），按时间倒序返回。日常调试 / 溯源用。 + + Args: + 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.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"} + + @mcp.tool() + def memory_atoms( + importance: str | None = None, + include_deprecated: bool = False, + limit: int = 50, + ) -> dict[str, Any]: + """**L2 原子记忆列表**：结构化查询原子记忆（按 importance 过滤）。 + + 与 ``memory_recall``（语义召回）互补——本工具做结构化枚举 + （importance / deprecated），返回原子断言 + 置信度 + 重要性。 + + Args: + importance: filter by importance (low/medium/high). + include_deprecated: include deprecated atoms (default False). + limit: max atoms (default 50). + """ + memory = _memory() + atoms = memory.list_atoms( + importance=importance, + include_deprecated=include_deprecated, + limit=limit, + ) + return { + "atoms": [ + { + "atom_id": a.id, + "assertion": a.assertion, + "entity_id": getattr(a, "entity_id", None), + "confidence": getattr(a, "confidence", None), + "importance": getattr(a, "importance", None), + "created_at": getattr(a, "created_at", None), + } + for a in atoms + ], + "count": len(atoms), + } + return mcp diff --git a/tests/unit/agents/test_memory_mcp.py b/tests/unit/agents/test_memory_mcp.py index 1be2924c..a360e98c 100644 --- a/tests/unit/agents/test_memory_mcp.py +++ b/tests/unit/agents/test_memory_mcp.py @@ -51,6 +51,12 @@ def test_build_registers_five_tools(fake_memory): "memory_capture", "memory_update", "memory_search_raw", + "memory_raws", + "memory_candidates", + "memory_extract", + "memory_promote", + "memory_reject", + "memory_atoms", } @@ -227,3 +233,43 @@ async def _send(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" From d629bb154333e51d04884ce32fa797c378dcb237 Mon Sep 17 00:00:00 2001 From: "Arvin.qi" Date: Wed, 26 Aug 2026 22:38:05 +0800 Subject: [PATCH 6/8] =?UTF-8?q?refactor(memory-mcp):=20=E7=B2=BE=E7=AE=80?= =?UTF-8?q?=E5=B7=A5=E5=85=B7=E9=9D=A2=E2=80=94=E2=80=94=E5=90=88=E5=B9=B6?= =?UTF-8?q?=20L0/L2=20=E6=9F=A5=E8=AF=A2=EF=BC=8C11=E2=86=929=20=E4=B8=AA?= =?UTF-8?q?=E5=B7=A5=E5=85=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 删除 memory_atoms（memory_recall 已覆盖 L2 查询） - 删除 memory_search_raw（memory_raws 增加 query 参数走 FTS，覆盖 L0 搜索） - 保留 9 个：recall/save/capture/update（读写）+ raws/candidates/extract/promote/reject（分层查询+流水线调度） --- src/octop/infra/agents/memory_mcp.py | 93 +++------------------------- tests/unit/agents/test_memory_mcp.py | 7 +-- 2 files changed, 13 insertions(+), 87 deletions(-) diff --git a/src/octop/infra/agents/memory_mcp.py b/src/octop/infra/agents/memory_mcp.py index 2416e3d7..59b1697a 100644 --- a/src/octop/infra/agents/memory_mcp.py +++ b/src/octop/infra/agents/memory_mcp.py @@ -14,7 +14,7 @@ * ``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_search_raw``; ``memory_recall`` returns it only + 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 @@ -225,7 +225,7 @@ def memory_capture( **日常使用**：把对话/事件原始内容记录下来，交给自动提取流水线 （extract -> candidate -> promote -> atom），稍后经 ``memory_recall`` - 可召回。记录后立即可用 ``memory_search_raw`` 查询。The source marker + 可召回。记录后立即可用 ``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. @@ -272,51 +272,12 @@ def memory_capture( "recorded": True, "extract_scheduled": extract_scheduled, "note": ( - "raw (L0) event recorded; visible now via memory_search_raw, " + "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_search_raw( - query: str, - limit: int = 10, - user: str | None = None, - ctx: Context | None = None, # type: ignore[type-arg] - ) -> dict[str, Any]: - """FTS-search L0 raw events of this expert (capture visible immediately). - - Unlike ``memory_recall`` (which reads atoms), this searches the raw - event layer, so records written by ``memory_capture`` are visible right - away, before extraction promotes them. - - Args: - query: keywords to match against raw event content. - limit: max number of events to return. - user: optional caller id (overrides the ``X-Octop-User-Id`` header); - returned in ``caller`` for per-caller traceability. - ctx: injected MCP context (reads ``X-Octop-User-Id`` header). - """ - caller = user or _caller_user(ctx) - memory = _memory() - events = memory.search_raw(query, limit=limit) - return { - "events": [ - { - "event_id": e.id, - "timestamp": e.timestamp.isoformat(), - "session_id": e.session_id, - "user": e.user, - "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_update( atom_id: str, @@ -366,18 +327,21 @@ def memory_update( @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 原始事件列表**：按 session/host/user 过滤查询原始事件（证据源）。 + """**L0 原始事件查询**：FTS 搜索或结构化过滤原始事件（证据源）。 - 与 ``memory_search_raw``（全文搜索）互补——本工具做结构化过滤 - （session/host/user），按时间倒序返回。日常调试 / 溯源用。 + 合并了原 ``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). @@ -386,7 +350,7 @@ def memory_raws( """ caller = user or _caller_user(ctx) memory = _memory() - events = memory.list_raw( + 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), @@ -541,43 +505,6 @@ def memory_reject( ) return {"ok": ok, "candidate_id": candidate_id, "status": "rejected"} - @mcp.tool() - def memory_atoms( - importance: str | None = None, - include_deprecated: bool = False, - limit: int = 50, - ) -> dict[str, Any]: - """**L2 原子记忆列表**：结构化查询原子记忆（按 importance 过滤）。 - - 与 ``memory_recall``（语义召回）互补——本工具做结构化枚举 - （importance / deprecated），返回原子断言 + 置信度 + 重要性。 - - Args: - importance: filter by importance (low/medium/high). - include_deprecated: include deprecated atoms (default False). - limit: max atoms (default 50). - """ - memory = _memory() - atoms = memory.list_atoms( - importance=importance, - include_deprecated=include_deprecated, - limit=limit, - ) - return { - "atoms": [ - { - "atom_id": a.id, - "assertion": a.assertion, - "entity_id": getattr(a, "entity_id", None), - "confidence": getattr(a, "confidence", None), - "importance": getattr(a, "importance", None), - "created_at": getattr(a, "created_at", None), - } - for a in atoms - ], - "count": len(atoms), - } - return mcp diff --git a/tests/unit/agents/test_memory_mcp.py b/tests/unit/agents/test_memory_mcp.py index a360e98c..9985dc30 100644 --- a/tests/unit/agents/test_memory_mcp.py +++ b/tests/unit/agents/test_memory_mcp.py @@ -50,13 +50,11 @@ def test_build_registers_five_tools(fake_memory): "memory_save", "memory_capture", "memory_update", - "memory_search_raw", "memory_raws", "memory_candidates", "memory_extract", "memory_promote", "memory_reject", - "memory_atoms", } @@ -105,18 +103,19 @@ def test_memory_capture_goes_add_raw(fake_memory): assert "raw (L0)" in result["note"] -def test_memory_search_raw_queries_l0(fake_memory): +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_search_raw"].fn(query="report panel banner", limit=5) + 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" From a857871bacc7a966b89b5c3d050e0e4d14b872ab Mon Sep 17 00:00:00 2001 From: "Arvin.qi" Date: Wed, 26 Aug 2026 22:48:30 +0800 Subject: [PATCH 7/8] chore: remove upstream PR.md draft (PR description lives on GitHub) --- PR.md | 113 ---------------------------------------------------------- 1 file changed, 113 deletions(-) delete mode 100644 PR.md diff --git a/PR.md b/PR.md deleted file mode 100644 index d843eb44..00000000 --- a/PR.md +++ /dev/null @@ -1,113 +0,0 @@ -# PR Title - -feat(memory): expose expert memory as an MCP server for external agents - ---- - -## Summary - -Adds a memory **MCP server** (Streamable HTTP at `/mcp/memory`) so external -agents (coding agents, bots, other AI tools) can directly **read / write / -update Octop expert memory**, aligned 1:1 with the in-process -`MemoryService` capabilities. Every write stamps a `source` marker that is -traceable on recall. - -## Why - -Octop experts accumulate rich memory (facts, conversations, decisions), but -today only the Octop dashboard / in-process agent can access it. External -agents that need to reuse that expertise (e.g. a coding agent asking a -business expert's accumulated knowledge) have no way in. This PR exposes the -same memory surface over the standard MCP protocol so any MCP-capable agent -can join the loop. - -## What - -- **New module** `src/octop/infra/agents/memory_mcp.py` — FastMCP server - bound to one expert per connection, plus token auth and header routing. -- **Mount** in `api/app.py` (`build_app`) at `/mcp/memory`, with - `streamable_http` task groups wired into the FastAPI lifespan. -- **Tests** `tests/unit/agents/test_memory_mcp.py` (13 tests). - -### Tools - -| Tool | Purpose | Backing API | -|------|---------|-------------| -| `memory_recall(query, limit=5)` | Recall memories (full pipeline: tokenize → FTS → rerank → dedupe); returns structured snippets + rendered markdown | `recall_for_prompt` | -| `memory_save(content, source, topic?)` | Persist a structured fact directly into the atom/tree (durable, no extraction) | `Memory.store` | -| `memory_capture(content, source, session_id?)` | Write an **L0 raw event** (goes through extraction); visible immediately via `memory_search_raw` | `Memory.add_raw` | -| `memory_search_raw(query, limit=10)` | FTS-search L0 raw events (capture visible before extraction) | `Memory.search_raw` | -| `memory_update(atom_id, new_content, source)` | Deprecate old atom + persist the new fact | `deprecate_atom` + `store` | - -### Expert binding & auth - -- **One connection binds one expert**: endpoint is a single `/mcp/memory`; - the expert is selected at connect time via the `X-Octop-Agent-Id` header — - callers never pass an agent id per tool call (they don't know the id list). -- **Auth**: independent token via `OCTOP_MEMORY_MCP_TOKEN` (fail-closed when - unset). Authorization via `Authorization: Bearer` or `X-Octop-Memory-Token`. - -### raw vs atom (for callers) - -- `memory_capture` → **L0 raw event** (evidence layer), distilled later by - the extraction pipeline (`extract → candidate → promote → atom`). Use it to - record raw conversations/events; the record is visible immediately via - `memory_search_raw` and recallable via `memory_recall` once promoted. -- `memory_save` → **atom/tree directly** (durable, no extraction). Use it - when the fact is already known. - -## Implementation notes - -- Lives in `infra/agents/` with no api-layer dependency: opens the agent - `Memory` instance via `open_memory_kwargs` + `Memory(...)` (workspace - resolved from the agent registry). -- DNS rebinding protection disabled (`TransportSecuritySettings`) because - Octop runs behind a reverse proxy (Host is the public domain, not localhost). -- `streamable_http_path` collapsed to `/` so the endpoint is exactly - `/mcp/memory` (the SDK default `/mcp` would yield `/mcp/memory/mcp`). -- One `FastMCP` per expert, routed by an ASGI dispatcher on the - `X-Octop-Agent-Id` header; missing/unknown agent → 404. - -## Usage example - -```bash -export OCTOP_MEMORY_MCP_TOKEN="" -``` - -```json -{ - "mcpServers": { - "octop-memory": { - "type": "streamable_http", - "url": "http:///mcp/memory/", - "headers": { - "Authorization": "Bearer ", - "X-Octop-Agent-Id": "" - } - } - } -} -``` - -```text -memory_recall(query="what are the key project decisions?") -memory_save(content="the release window is every Tuesday", source="coding-agent", topic="release") -memory_capture(content="user reported: the report panel banner is not rendering", source="review-bot", session_id="review-2026-08-20") -memory_search_raw(query="report panel banner") -memory_update(atom_id="atom_xxx", new_content="updated fact", source="coding-agent") -``` - -## Testing - -- `tests/unit/agents/test_memory_mcp.py` — 13 tests: tool registration, - recall pipeline, capture (raw) semantics, search_raw, update, token - middleware (401 / accept), header routing, 404 unknown agent, unified mount. -- Verified locally by booting the server and exercising the MCP endpoints: - health, 401 without token, `initialize` (binds expert via header), - `tools/list` (5 tools), `tools/call memory_recall`. - -## Checklist - -- [x] No internal/hard-coded environment-specific values in the diff -- [x] `make lint` clean (ruff) -- [x] Unit tests pass From 97f1eb9b04692f22c64828b96775887596229f2c Mon Sep 17 00:00:00 2001 From: jinlongqi Date: Fri, 28 Aug 2026 14:50:49 +0800 Subject: [PATCH 8/8] =?UTF-8?q?docs(memory-mcp):=20=E6=96=B0=E5=A2=9E?= =?UTF-8?q?=E4=B8=93=E5=AE=B6=E8=AE=B0=E5=BF=86=20MCP=20Server=20=E4=BD=BF?= =?UTF-8?q?=E7=94=A8=E6=96=87=E6=A1=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 覆盖 9 工具面（recall/save/capture/update/raws/candidates/extract/promote/reject）： - 连接信息：/mcp/memory endpoint、OCTOP_MEMORY_MCP_TOKEN 鉴权（fail-closed）、X-Octop-Agent-Id 专家绑定、调用者追溯 - 工具参数表 + 与 MemoryService 的层级映射（L0/L1/L2） - 使用示例：curl 握手、capture/save/recall、MCP SDK 客户端 - 部署配置与开启步骤 --- docs/memory-mcp.md | 191 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 191 insertions(+) create mode 100644 docs/memory-mcp.md 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。