最近在调研 Agent 长期记忆方案时发现很多团队一上来就考虑向量数据库、图数据库这些重型存储结果架构做得很复杂日常开发中却连“记住用户上一个需求”这种基础问题都还没解决好。其实大部分 Agent 记忆场景用合适的数据结构加全文检索就能覆盖绝大多数需求。Show HN 上这个 MCP Memory 项目就很有代表性——它用 Google 的 OKF 做内容指纹与去重用 SQLite 内置的 FTS5 做快速记忆检索最后把能力封装成 MCP Server暴露给 Claude Desktop、Codex、Cline 这类 AI 客户端使用。本文会围绕这个思路从概念讲到源码再带大家从零实现一个可运行的 MCP Memory Server包含数据库设计、工具封装、客户端配置和排错指南。无论你是刚开始接触 MCP还是已经在做 Agent 应用这篇文章都能给你一套可以直接落地的项目模板。1. 为什么 Agent 需要记忆从一次对话说起1.1 Agent 的“失忆”困境先想象一个场景你正在用 Claude Desktop 或 Cline 写代码你和 AI 约定好“项目里所有配置项统一放到 config/ 目录下”。前几分钟 AI 还能遵守这个约定但一旦对话上下文超出窗口或者你开了新会话它就把这个约定忘得一干二净。下一次它可能又把配置写在根目录你要重新解释一遍。这就是 Agent 的“失忆”问题。大模型的对话上下文是有长度限制的ChatGPT、Claude、Codex 这类产品虽然不断在扩大上下文窗口但我们不可能把所有历史信息都塞进 context 里不仅成本高而且信息一多模型反而抓不住重点。所以我们需要一个“外部记忆层”短期记忆当前会话里正在讨论的内容一般存在于上下文窗口。长期记忆跨会话保留的事实、偏好、决策记录、项目约定需要持久化存储和检索。工作记忆任务执行过程中的临时状态比如正在处理哪个文件、下一步要做什么。长期记忆正是 MCP Memory 这类项目要解决的问题。它需要做到三件事能写入能去重能快速检索。1.2 常见 Agent Memory 方案对比现有方案大致可以分为几类方案优点缺点向量数据库如 Chroma、Qdrant语义检索效果好适合模糊匹配部署重、需要 embedding 模型本地开发成本高图数据库Neo4j 等关系表达能力强查询复杂维护成本高Redis / KV 存储读写快只能按 key 查无法做内容级检索关系型数据库 LIKE实现简单性能差不支持相关度排序SQLite FTS5 全文检索轻量、零部署、检索快、支持 BM25 排序语义检索能力弱需配合分词技巧MCP Memory 选择的是最后一种SQLite 作为存储底座FTS5 负责全文搜索。它不需要额外起服务不需要 embedding 模型一个文件就是一个数据库非常适合本地开发、单机部署、轻量级工具类 Agent 场景。1.3 MCP 在这张图里的位置你可能已经接触过 MCPModel Context Protocol模型上下文协议。简单来说它是由 Anthropic 提出并推动的开放协议用来标准化大模型应用与外部工具、数据源之间的交互方式。MCP Server 可以暴露三类能力Tools可被模型调用的函数比如 remember、recall。Resources可被模型读取的数据资源比如记忆列表、配置文档。Prompts可复用的提示词模板用于引导模型完成特定流程。本文实现的 MCP Memory本质就是一个 MCP Server对外提供记忆的写入、检索、删除等 Tools以及最近记忆等 Resources。这里顺便区分一个高频问题Agent Skill 和 MCP 有什么区别Skill 更像是对 Agent 能力的描述和提示词编排告诉模型“你会做什么、应该怎么做”MCP 则是标准化的协议通道解决的是“模型如何调用外部工具/数据”的通信问题。两者可以搭配使用Skill 定义行为MCP 提供执行通道。2. 环境准备与版本说明在写代码之前先把环境准备好。2.1 运行环境本文示例以本地开发环境为主条件如下操作系统macOS / Linux / Windows 均可示例代码全部跨平台。Python 版本3.10 或以上3.11 更推荐。SQLite 版本建议 3.34.0 以上因为我们要使用 FTS5 的 trigram tokenizer 来优化中文检索。MCP Python SDK需要安装mcp包。先检查本地 Python 和 SQLite 版本python3 --version python3 -c import sqlite3; print(sqlite3.sqlite_version)如果 SQLite 版本低于 3.34建议升级 Python 或使用 pysqlite3后面会在常见问题里专门说明。2.2 安装 MCP SDK使用 pip 安装pip install mcp这个包会提供FastMCP、ClientSession、stdio_client等核心组件我们可以用它快速构建 MCP Server 和测试客户端。2.3 项目结构我们先规划一个清晰的目录结构方便后续扩展mcp-memory/ ├── mcp_memory_server.py # MCP Server 主程序 ├── memory.db # SQLite 数据库文件运行时自动生成 ├── test_client.py # 本地调试客户端 └── requirements.txt # 依赖列表requirements.txt内容mcp1.0.0如果你需要安装到虚拟环境可以执行python3 -m venv .venv source .venv/bin/activate # Windows 下用 .venv\Scripts\activate pip install -r requirements.txt3. 核心原理拆解OKF 与 SQLite FTS5 能做什么3.1 OKF给记忆内容生成“稳定指纹”OKF 全称 Object Key Fingerprinting是 Google 开源的一套内容指纹思路核心目标是给对象内容生成稳定、跨语言可复现的指纹fingerprint用于内容寻址和去重。为什么要用它来做 Agent Memory因为记忆最怕重复写入。用户多次告诉 AI “我喜欢用 Python”每次措辞略有不同但本质是一回事。如果每次都新写一条记录记忆库会迅速膨胀检索时还会返回大量重复内容降低召回质量。OKF 的价值就在于我们把记忆内容规范化成 JSON 对象后计算出一个固定长度的指纹用这个指纹作为唯一键。下次写入相同内容时直接走更新逻辑而不是新增记录。实际项目中可以直接使用google/okf的 Python SDK。不过它的具体 API 版本和项目实现绑定较紧为了本文示例的可运行性我们先用 Python 标准库的hashlib.blake2b实现一个“OKF 风格”的指纹函数思路完全一致但你需要根据实际依赖调整成官方 SDK 调用。import hashlib import json def okf_fingerprint(obj: object) - str: 将任意对象规范化为 JSON 字符串再计算 BLAKE2b 指纹。 关键点 1. sort_keysTrue 保证字段顺序不影响指纹。 2. separators(,, :) 去掉多余空格保证跨环境一致。 3. ensure_asciiFalse 保留中文原文避免编码差异。 canonical json.dumps( obj, sort_keysTrue, separators(,, :), ensure_asciiFalse, ) return hashlib.blake2b(canonical.encode(utf-8), digest_size16).hexdigest()在实际项目中如果你使用google/okf官方 SDK它通常有自己的 ObjectKey 生成与验证逻辑性能上也可能有针对性优化。本文核心是讲清楚“为什么要指纹”以及“指纹用于哪个表字段”代码思路可以直接迁移。3.2 SQLite FTS5轻量级全文搜索引擎FTS5 是 SQLite 的全文搜索扩展模块不需要独立服务不需要额外文件直接内嵌在 SQLite 引擎里。它支持BM25 相关度排序结果按相关度从高到低返回。前缀查询、短语查询、AND / OR / NOT 逻辑操作。外部内容表external content table可以关联到普通业务表。多种 tokenizer其中 trigram 支持子串匹配对中文比较友好。FTS5 的经典用法是创建一个虚拟表虚拟表内部维护倒排索引。写入数据时我们同时维护业务表和 FTS 索引查询时直接用 MATCH 语法。3.3 基于 external content 表的设计这是 FTS5 的推荐用法FTS 虚拟表只做索引不存真实内容真实内容仍然存在普通表里。好处是不需要冗余存储大段文本且业务表增删改时用触发器同步索引。设计如下memories主表存记忆内容、类型、标签、时间等字段。memories_ftsFTS5 虚拟表通过contentmemories关联主表。三个触发器插入、更新、删除时自动同步 FTS 索引。这样上层代码不用关心索引同步细节只要操作主表FTS 索引自动保持一致。4. 完整实战从零实现一个 MCP Memory Server接下来是本文的核心。我们分几步完成初始化数据库、实现指纹函数、封装 MCP 工具、编写本地测试客户端验证。4.1 初始化数据库在mcp_memory_server.py中先定义数据库路径和初始化函数。数据库路径建议通过环境变量MEMORY_DB_PATH指定这样客户端配置时可以灵活调整。import os DB_PATH os.environ.get(MEMORY_DB_PATH, os.path.join(os.path.dirname(__file__), memory.db)) def init_db(): 初始化数据库表结构和 FTS5 索引。 conn sqlite3.connect(DB_PATH) conn.executescript( CREATE TABLE IF NOT EXISTS memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, memo_id TEXT NOT NULL UNIQUE, okf TEXT NOT NULL UNIQUE, content TEXT NOT NULL, kind TEXT NOT NULL DEFAULT note, tags TEXT NOT NULL DEFAULT , source TEXT NOT NULL DEFAULT manual, created_at TEXT NOT NULL DEFAULT (datetime(now)), updated_at TEXT NOT NULL DEFAULT (datetime(now)) ); -- FTS5 外部内容表tokenizer 使用 trigram支持中文子串匹配 CREATE VIRTUAL TABLE IF NOT EXISTS memories_fts USING fts5( content, kind, tags, contentmemories, content_rowidid, tokenizetrigram ); -- 插入触发器向 FTS 表写入索引 CREATE TRIGGER IF NOT EXISTS memories_ai AFTER INSERT ON memories BEGIN INSERT INTO memories_fts(rowid, content, kind, tags) VALUES (new.id, new.content, new.kind, new.tags); END; -- 删除触发器删除 FTS 索引中的对应行 CREATE TRIGGER IF NOT EXISTS memories_ad AFTER DELETE ON memories BEGIN INSERT INTO memories_fts(memories_fts, rowid, content, kind, tags) VALUES (delete, old.id, old.content, old.kind, old.tags); END; -- 更新触发器先删旧索引再插入新索引 CREATE TRIGGER IF NOT EXISTS memories_au AFTER UPDATE ON memories BEGIN INSERT INTO memories_fts(memories_fts, rowid, content, kind, tags) VALUES (delete, old.id, old.content, old.kind, old.tags); INSERT INTO memories_fts(rowid, content, kind, tags) VALUES (new.id, new.content, new.kind, new.tags); END; ) conn.commit() conn.close()这里的重点是memo_id是对外暴露的记忆 ID使用 UUID 字符串方便客户端引用。okf是内容指纹做唯一去重。memories_fts使用contentmemories和content_rowididFTS 索引直接关联主表的 rowid。触发器保证索引同步UPDATE触发器先delete再insert这是 FTS5 external content 表的官方推荐写法。再写一个获取数据库连接的辅助函数def get_conn() - sqlite3.Connection: conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row conn.execute(PRAGMA journal_modeWAL;) conn.execute(PRAGMA busy_timeout5000;) return conn启用 WAL 模式之后SQLite 支持读写并发适合 MCP Server 多会话场景。busy_timeout可以避免并发写入时直接报 “database is locked”。4.2 实现 MCP Server 入口使用 FastMCP 创建 Serverimport uuid from mcp.server.fastmcp import FastMCP mcp FastMCP(mcp-memory)4.3 实现记忆写入工具rememberremember工具接受内容、类型、标签、来源等参数。逻辑如下基于 content 和 kind 计算 OKF 指纹。查询数据库是否已存在该指纹。新增或更新记录。返回操作结果。mcp.tool() def remember( content: str, kind: str note, tags: str , source: str manual, ) - dict: 记录一条新记忆如果内容指纹相同则更新已有记录。 Args: content: 记忆内容例如“用户喜欢用 Python 写后端”。 kind: 记忆类型如 preference、fact、decision、note。 tags: 逗号分隔的标签例如“python, fastapi”。 source: 记忆来源例如 manual、claude、codex。 content content.strip() if not content: return {status: error, message: content cannot be empty} fp okf_fingerprint({content: content, kind: kind}) conn get_conn() existing conn.execute( SELECT memo_id FROM memories WHERE okf ?, (fp,) ).fetchone() if existing: conn.execute( UPDATE memories SET tags ?, source ?, updated_at datetime(now) WHERE okf ? , (tags, source, fp), ) conn.commit() conn.close() return { status: updated, memo_id: existing[memo_id], okf: fp, } memo_id str(uuid.uuid4()) conn.execute( INSERT INTO memories (memo_id, okf, content, kind, tags, source) VALUES (?, ?, ?, ?, ?, ?) , (memo_id, fp, content, kind, tags, source), ) conn.commit() conn.close() return { status: created, memo_id: memo_id, okf: fp, }这里的设计细节是指纹只对content kind计算tags 变化不算新记忆。这样做的好处是“内容一致但标签补全”时只是更新标签不会产生重复记忆。如果你希望 tags 变化也算新记忆可以把 tags 加进指纹对象按业务需要调整。4.4 实现记忆检索工具recallrecall是记忆系统最核心的读取接口。它通过 FTS5 全文检索实现对输入 query 做清洗拆成多个有效词。对长度 3 的词走 FTS5 MATCH词之间用 OR 连接保证召回率。如果 query 过短或为空退化为 LIKE 查询或直接返回最新记录。按 BM25 相关度升序排序FTS5 中 BM25 分数越低越相关。import re def build_match_query(query: str) - str: 把用户输入拆成 token并用 OR 连接成 FTS5 查询表达式。 trigram tokenizer 要求查询词至少 3 个字符所以这里过滤掉短词。 terms [] for token in re.split(r\s, query.strip()): token token.strip().strip() if len(token) 3: terms.append(f{token.replace(chr(34), chr(34) * 2)}) return OR .join(terms) mcp.tool() def recall(query: str, limit: int 10) - list[dict]: 根据关键词检索历史记忆按相关度排序。 Args: query: 关键词例如“用户喜欢什么编程语言”。 limit: 返回条数默认 10。 query query.strip() limit max(1, min(int(limit), 50)) conn get_conn() if len(query) 3: match_expr build_match_query(query) if match_expr: rows conn.execute( SELECT m.memo_id, m.content, m.kind, m.tags, m.source, m.created_at, m.updated_at, bm25(memories_fts) AS score FROM memories_fts JOIN memories m ON m.id memories_fts.rowid WHERE memories_fts MATCH ? ORDER BY score ASC LIMIT ? , (match_expr, limit), ).fetchall() conn.close() return [dict(row) for row in rows] # 退化为 LIKE 查询适合不足 3 个字符或空查询 like f%{query}% rows conn.execute( SELECT memo_id, content, kind, tags, source, created_at, updated_at, 0 AS score FROM memories WHERE content LIKE ? OR tags LIKE ? ORDER BY updated_at DESC LIMIT ? , (like, like, limit), ).fetchall() conn.close() return [dict(row) for row in rows]说明几点bm25(memories_fts)返回的分数越低代表相关度越高因此ORDER BY score ASC。为什么用 OR 而不是 AND记忆检索场景我们希望更“宽容”只要记忆包含任意一个关键词就应该召回再由 BM25 排序决定谁更靠前。如果用 AND稍微长一点的查询就很容易查不到结果。如果 query 很短比如 12 个字符trigram tokenizer 无法匹配我们直接退回 LIKE。4.5 实现删除与统计工具删除操作在生产环境必须谨慎。下面的forget工具先根据 memo_id 查存在性再执行删除实际应用中建议增加二次确认逻辑比如要求额外传入confirmTrue。mcp.tool() def forget(memo_id: str, confirm: bool False) - dict: 删除一条记忆。必须显式传入 confirmTrue 才会真正执行删除。 Args: memo_id: 记忆 ID由 remember 返回。 confirm: 是否确认删除必须为 True。 if not confirm: return {status: cancelled, message: 请传入 confirmTrue 确认删除} conn get_conn() existing conn.execute( SELECT memo_id FROM memories WHERE memo_id ?, (memo_id,) ).fetchone() if not existing: conn.close() return {status: not_found, memo_id: memo_id} conn.execute(DELETE FROM memories WHERE memo_id ?, (memo_id,)) conn.commit() conn.close() return {status: deleted, memo_id: memo_id}再加一个简单的统计工具方便后续做记忆质量分析和清理mcp.tool() def memory_stats() - dict: 返回记忆库的统计信息。 conn get_conn() total conn.execute(SELECT COUNT(*) AS total FROM memories).fetchone()[total] kinds conn.execute( SELECT kind, COUNT(*) AS cnt FROM memories GROUP BY kind ORDER BY cnt DESC ).fetchall() conn.close() return { total: total, by_kind: [dict(row) for row in kinds], }4.6 实现 Resources 和 Prompts除了 ToolsMCP Server 还可以暴露 Resources 和 Prompts让模型直接读取或按模板执行。mcp.resource(memory://recent) def recent_memories() - str: 返回最近 10 条记忆供模型直接读取。 conn get_conn() rows conn.execute( SELECT memo_id, content, kind, tags, updated_at FROM memories ORDER BY updated_at DESC LIMIT 10 ).fetchall() conn.close() lines [f- [{row[kind]}] {row[content]} (tags: {row[tags]}) for row in rows] return \n.join(lines) if lines else 暂无记忆。 mcp.prompt() def memory_retrieval(question: str) - str: 当用户提问时先检索历史记忆再回答。 return f请先调用 recall 工具检索与「{question}」相关的历史记忆再结合记忆内容回答用户。如果检索结果为空请直接说明没有找到相关记忆。4.7 启动入口最后加上启动入口if __name__ __main__: init_db() mcp.run(transportstdio)启动脚本完整文件如下路径mcp_memory_server.pyimport hashlib import json import os import re import sqlite3 import uuid from mcp.server.fastmcp import FastMCP DB_PATH os.environ.get(MEMORY_DB_PATH, os.path.join(os.path.dirname(__file__), memory.db)) mcp FastMCP(mcp-memory) def okf_fingerprint(obj: object) - str: canonical json.dumps( obj, sort_keysTrue, separators(,, :), ensure_asciiFalse, ) return hashlib.blake2b(canonical.encode(utf-8), digest_size16).hexdigest() def get_conn() - sqlite3.Connection: conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row conn.execute(PRAGMA journal_modeWAL;) conn.execute(PRAGMA busy_timeout5000;) return conn def init_db(): conn sqlite3.connect(DB_PATH) conn.executescript( CREATE TABLE IF NOT EXISTS memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, memo_id TEXT NOT NULL UNIQUE, okf TEXT NOT NULL UNIQUE, content TEXT NOT NULL, kind TEXT NOT NULL DEFAULT note, tags TEXT NOT NULL DEFAULT , source TEXT NOT NULL DEFAULT manual, created_at TEXT NOT NULL DEFAULT (datetime(now)), updated_at TEXT NOT NULL DEFAULT (datetime(now)) ); CREATE VIRTUAL TABLE IF NOT EXISTS memories_fts USING fts5( content, kind, tags, contentmemories, content_rowidid, tokenizetrigram ); CREATE TRIGGER IF NOT EXISTS memories_ai AFTER INSERT ON memories BEGIN INSERT INTO memories_fts(rowid, content, kind, tags) VALUES (new.id, new.content, new.kind, new.tags); END; CREATE TRIGGER IF NOT EXISTS memories_ad AFTER DELETE ON memories BEGIN INSERT INTO memories_fts(memories_fts, rowid, content, kind, tags) VALUES (delete, old.id, old.content, old.kind, old.tags); END; CREATE TRIGGER IF NOT EXISTS memories_au AFTER UPDATE ON memories BEGIN INSERT INTO memories_fts(memories_fts, rowid, content, kind, tags) VALUES (delete, old.id, old.content, old.kind, old.tags); INSERT INTO memories_fts(rowid, content, kind, tags) VALUES (new.id, new.content, new.kind, new.tags); END; ) conn.commit() conn.close() def build_match_query(query: str) - str: terms [] for token in re.split(r\s, query.strip()): token token.strip().strip() if len(token) 3: terms.append(f{token.replace(chr(34), chr(34) * 2)}) return OR .join(terms) mcp.tool() def remember(content: str, kind: str note, tags: str , source: str manual) - dict: content content.strip() if not content: return {status: error, message: content cannot be empty} fp okf_fingerprint({content: content, kind: kind}) conn get_conn() existing conn.execute( SELECT memo_id FROM memories WHERE okf ?, (fp,) ).fetchone() if existing: conn.execute( UPDATE memories SET tags ?, source ?, updated_at datetime(now) WHERE okf ? , (tags, source, fp), ) conn.commit() conn.close() return {status: updated, memo_id: existing[memo_id], okf: fp} memo_id str(uuid.uuid4()) conn.execute( INSERT INTO memories (memo_id, okf, content, kind, tags, source) VALUES (?, ?, ?, ?, ?, ?) , (memo_id, fp, content, kind, tags, source), ) conn.commit() conn.close() return {status: created, memo_id: memo_id, okf: fp} mcp.tool() def recall(query: str, limit: int 10) - list[dict]: query query.strip() limit max(1, min(int(limit), 50)) conn get_conn() if len(query) 3: match_expr build_match_query(query) if match_expr: rows conn.execute( SELECT m.memo_id, m.content, m.kind, m.tags, m.source, m.created_at, m.updated_at, bm25(memories_fts) AS score FROM memories_fts JOIN memories m ON m.id memories_fts.rowid WHERE memories_fts MATCH ? ORDER BY score ASC LIMIT ? , (match_expr, limit), ).fetchall() conn.close() return [dict(row) for row in rows] like f%{query}% rows conn.execute( SELECT memo_id, content, kind, tags, source, created_at, updated_at, 0 AS score FROM memories WHERE content LIKE ? OR tags LIKE ? ORDER BY updated_at DESC LIMIT ? , (like, like, limit), ).fetchall() conn.close() return [dict(row) for row in rows] mcp.tool() def forget(memo_id: str, confirm: bool False) - dict: if not confirm: return {status: cancelled, message: 请传入 confirmTrue 确认删除} conn get_conn() existing conn.execute( SELECT memo_id FROM memories WHERE memo_id ?, (memo_id,) ).fetchone() if not existing: conn.close() return {status: not_found, memo_id: memo_id} conn.execute(DELETE FROM memories WHERE memo_id ?, (memo_id,)) conn.commit() conn.close() return {status: deleted, memo_id: memo_id} mcp.tool() def memory_stats() - dict: conn get_conn() total conn.execute(SELECT COUNT(*) AS total FROM memories).fetchone()[total] kinds conn.execute( SELECT kind, COUNT(*) AS cnt FROM memories GROUP BY kind ORDER BY cnt DESC ).fetchall() conn.close() return {total: total, by_kind: [dict(row) for row in kinds]} mcp.resource(memory://recent) def recent_memories() - str: conn get_conn() rows conn.execute( SELECT memo_id, content, kind, tags, updated_at FROM memories ORDER BY updated_at DESC LIMIT 10 ).fetchall() conn.close() lines [f- [{row[kind]}] {row[content]} (tags: {row[tags]}) for row in rows] return \n.join(lines) if lines else 暂无记忆。 mcp.prompt() def memory_retrieval(question: str) - str: return f请先调用 recall 工具检索与「{question}」相关的历史记忆再结合记忆内容回答用户。如果检索结果为空请直接说明没有找到相关记忆。 if __name__ __main__: init_db() mcp.run(transportstdio)4.8 编写测试客户端验证由于 MCP Server 是基于 stdio 传输的直接在命令行里不好模拟双向通信。我建议写一个简单的测试客户端用 MCP Python SDK 连接并调用工具。# 文件路径test_client.py import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): server_params StdioServerParameters( commandpython, args[mcp_memory_server.py], env{MEMORY_DB_PATH: memory.db}, ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() print(可用工具, [tool.name for tool in tools.tools]) res await session.call_tool( remember, arguments{ content: 用户喜欢用 Python 和 FastAPI 写后端服务, kind: preference, tags: python, fastapi, source: test, }, ) print(remember 结果, res.content) res await session.call_tool( recall, arguments{query: 用户喜欢什么语言, limit: 5}, ) print(recall 结果, res.content) if __name__ __main__: asyncio.run(main())运行方式python test_client.py预期会输出类似下面的内容可用工具 [remember, recall, forget, memory_stats] remember 结果 [TextContent(typetext, text{status: created, memo_id: xxxx, okf: xxxx})] recall 结果 [TextContent(typetext, text[{memo_id: xxxx, content: 用户喜欢用 Python 和 FastAPI 写后端服务, ...}])]到这里一个功能完整的 MCP Memory Server 已经跑通了。但你可能注意到一点我们目前还没有真正让 Claude、Codex、Cline 这些客户端使用它。下面就来解决接入问题。5. 将 MCP Server 接入客户端有了 MCP Server下一步就是把服务注册到 AI 客户端里。不同客户端的配置方式略有差异但核心思路一致指定command、args、env让客户端通过 stdio 拉起我们的 Server。5.1 接入 Claude DesktopClaude Desktop 的配置文件路径macOS~/Library/Application Support/Claude/claude_desktop_config.jsonWindows%APPDATA%\Claude\claude_desktop_config.json在配置文件的mcpServers中新增一项{ mcpServers: { mcp-memory: { command: python, args: [ /绝对路径/mcp_memory_server.py ], env: { MEMORY_DB_PATH: /绝对路径/memory.db } } } }保存后重启 Claude Desktop然后在对话中输入“帮我把这条记忆记下来用户喜欢用 Python”模型就会调用remember工具。新会话里问“我平时喜欢用什么语言”模型会调用recall工具从数据库里把记忆找回来。这里有一个很重要的坑command里的python必须能直接在客户端环境里启动。如果你是用虚拟环境安装的 mcp 包建议把command改成虚拟环境里的绝对路径例如/path/to/.venv/bin/pythonWindows 下是.venv\Scripts\python.exe否则客户端找不到依赖包。5.2 接入 Codex 与 ClineCodex 是 OpenAI 的命令行编程工具支持通过--mcp-config指定 MCP 配置文件。不同版本对配置文件的格式要求略有差异建议先查询你当前版本的支持情况。一般形式是codex --mcp-config mcp.json其中mcp.json内容类似{ mcpServers: { mcp-memory: { command: python, args: [/绝对路径/mcp_memory_server.py], env: { MEMORY_DB_PATH: /绝对路径/memory.db } } } }Cline 是 VS Code 里常用的 AI 编程插件它的 MCP Server 配置界面在插件设置中。添加方式打开 Cline 设置。找到 MCP Servers 配置。选择类型为stdio。Command 填python。Args 填[/绝对路径/mcp_memory_server.py]。Environment 里设置MEMORY_DB_PATH。注意如果你在 Codex、Cline 这类工具里发现“工具注册不上”常见原因有三个一是脚本启动时报错导致注册失败二是 Python 环境不对三是 Server 初始化耗时过长客户端超时。建议先用下面的 Inspector 方式调试。5.3 用 MCP Inspector 快速调试MCP Python SDK 提供了 Inspector 调试工具可以可视化查看 Server 暴露了哪些工具、参数是什么。运行方式mcp dev mcp_memory_server.py如果你的 SDK 版本支持该命令它会启动一个本地 Web 面板你可以在面板里看到工具列表手动调用并观察返回结果排查问题比直接接客户端快很多。6. 常见问题与排查思路问题现象常见原因解决思路启动时报no such module: fts5Python 自带的 sqlite3 没有启用 FTS5升级 Python 版本或安装 pysqlite3 并替换 sqlite3 模块客户端显示工具注册失败脚本启动
基于SQLite FTS5与OKF的MCP Memory Server实战：为Agent构建轻量级长期记忆
最近在调研 Agent 长期记忆方案时发现很多团队一上来就考虑向量数据库、图数据库这些重型存储结果架构做得很复杂日常开发中却连“记住用户上一个需求”这种基础问题都还没解决好。其实大部分 Agent 记忆场景用合适的数据结构加全文检索就能覆盖绝大多数需求。Show HN 上这个 MCP Memory 项目就很有代表性——它用 Google 的 OKF 做内容指纹与去重用 SQLite 内置的 FTS5 做快速记忆检索最后把能力封装成 MCP Server暴露给 Claude Desktop、Codex、Cline 这类 AI 客户端使用。本文会围绕这个思路从概念讲到源码再带大家从零实现一个可运行的 MCP Memory Server包含数据库设计、工具封装、客户端配置和排错指南。无论你是刚开始接触 MCP还是已经在做 Agent 应用这篇文章都能给你一套可以直接落地的项目模板。1. 为什么 Agent 需要记忆从一次对话说起1.1 Agent 的“失忆”困境先想象一个场景你正在用 Claude Desktop 或 Cline 写代码你和 AI 约定好“项目里所有配置项统一放到 config/ 目录下”。前几分钟 AI 还能遵守这个约定但一旦对话上下文超出窗口或者你开了新会话它就把这个约定忘得一干二净。下一次它可能又把配置写在根目录你要重新解释一遍。这就是 Agent 的“失忆”问题。大模型的对话上下文是有长度限制的ChatGPT、Claude、Codex 这类产品虽然不断在扩大上下文窗口但我们不可能把所有历史信息都塞进 context 里不仅成本高而且信息一多模型反而抓不住重点。所以我们需要一个“外部记忆层”短期记忆当前会话里正在讨论的内容一般存在于上下文窗口。长期记忆跨会话保留的事实、偏好、决策记录、项目约定需要持久化存储和检索。工作记忆任务执行过程中的临时状态比如正在处理哪个文件、下一步要做什么。长期记忆正是 MCP Memory 这类项目要解决的问题。它需要做到三件事能写入能去重能快速检索。1.2 常见 Agent Memory 方案对比现有方案大致可以分为几类方案优点缺点向量数据库如 Chroma、Qdrant语义检索效果好适合模糊匹配部署重、需要 embedding 模型本地开发成本高图数据库Neo4j 等关系表达能力强查询复杂维护成本高Redis / KV 存储读写快只能按 key 查无法做内容级检索关系型数据库 LIKE实现简单性能差不支持相关度排序SQLite FTS5 全文检索轻量、零部署、检索快、支持 BM25 排序语义检索能力弱需配合分词技巧MCP Memory 选择的是最后一种SQLite 作为存储底座FTS5 负责全文搜索。它不需要额外起服务不需要 embedding 模型一个文件就是一个数据库非常适合本地开发、单机部署、轻量级工具类 Agent 场景。1.3 MCP 在这张图里的位置你可能已经接触过 MCPModel Context Protocol模型上下文协议。简单来说它是由 Anthropic 提出并推动的开放协议用来标准化大模型应用与外部工具、数据源之间的交互方式。MCP Server 可以暴露三类能力Tools可被模型调用的函数比如 remember、recall。Resources可被模型读取的数据资源比如记忆列表、配置文档。Prompts可复用的提示词模板用于引导模型完成特定流程。本文实现的 MCP Memory本质就是一个 MCP Server对外提供记忆的写入、检索、删除等 Tools以及最近记忆等 Resources。这里顺便区分一个高频问题Agent Skill 和 MCP 有什么区别Skill 更像是对 Agent 能力的描述和提示词编排告诉模型“你会做什么、应该怎么做”MCP 则是标准化的协议通道解决的是“模型如何调用外部工具/数据”的通信问题。两者可以搭配使用Skill 定义行为MCP 提供执行通道。2. 环境准备与版本说明在写代码之前先把环境准备好。2.1 运行环境本文示例以本地开发环境为主条件如下操作系统macOS / Linux / Windows 均可示例代码全部跨平台。Python 版本3.10 或以上3.11 更推荐。SQLite 版本建议 3.34.0 以上因为我们要使用 FTS5 的 trigram tokenizer 来优化中文检索。MCP Python SDK需要安装mcp包。先检查本地 Python 和 SQLite 版本python3 --version python3 -c import sqlite3; print(sqlite3.sqlite_version)如果 SQLite 版本低于 3.34建议升级 Python 或使用 pysqlite3后面会在常见问题里专门说明。2.2 安装 MCP SDK使用 pip 安装pip install mcp这个包会提供FastMCP、ClientSession、stdio_client等核心组件我们可以用它快速构建 MCP Server 和测试客户端。2.3 项目结构我们先规划一个清晰的目录结构方便后续扩展mcp-memory/ ├── mcp_memory_server.py # MCP Server 主程序 ├── memory.db # SQLite 数据库文件运行时自动生成 ├── test_client.py # 本地调试客户端 └── requirements.txt # 依赖列表requirements.txt内容mcp1.0.0如果你需要安装到虚拟环境可以执行python3 -m venv .venv source .venv/bin/activate # Windows 下用 .venv\Scripts\activate pip install -r requirements.txt3. 核心原理拆解OKF 与 SQLite FTS5 能做什么3.1 OKF给记忆内容生成“稳定指纹”OKF 全称 Object Key Fingerprinting是 Google 开源的一套内容指纹思路核心目标是给对象内容生成稳定、跨语言可复现的指纹fingerprint用于内容寻址和去重。为什么要用它来做 Agent Memory因为记忆最怕重复写入。用户多次告诉 AI “我喜欢用 Python”每次措辞略有不同但本质是一回事。如果每次都新写一条记录记忆库会迅速膨胀检索时还会返回大量重复内容降低召回质量。OKF 的价值就在于我们把记忆内容规范化成 JSON 对象后计算出一个固定长度的指纹用这个指纹作为唯一键。下次写入相同内容时直接走更新逻辑而不是新增记录。实际项目中可以直接使用google/okf的 Python SDK。不过它的具体 API 版本和项目实现绑定较紧为了本文示例的可运行性我们先用 Python 标准库的hashlib.blake2b实现一个“OKF 风格”的指纹函数思路完全一致但你需要根据实际依赖调整成官方 SDK 调用。import hashlib import json def okf_fingerprint(obj: object) - str: 将任意对象规范化为 JSON 字符串再计算 BLAKE2b 指纹。 关键点 1. sort_keysTrue 保证字段顺序不影响指纹。 2. separators(,, :) 去掉多余空格保证跨环境一致。 3. ensure_asciiFalse 保留中文原文避免编码差异。 canonical json.dumps( obj, sort_keysTrue, separators(,, :), ensure_asciiFalse, ) return hashlib.blake2b(canonical.encode(utf-8), digest_size16).hexdigest()在实际项目中如果你使用google/okf官方 SDK它通常有自己的 ObjectKey 生成与验证逻辑性能上也可能有针对性优化。本文核心是讲清楚“为什么要指纹”以及“指纹用于哪个表字段”代码思路可以直接迁移。3.2 SQLite FTS5轻量级全文搜索引擎FTS5 是 SQLite 的全文搜索扩展模块不需要独立服务不需要额外文件直接内嵌在 SQLite 引擎里。它支持BM25 相关度排序结果按相关度从高到低返回。前缀查询、短语查询、AND / OR / NOT 逻辑操作。外部内容表external content table可以关联到普通业务表。多种 tokenizer其中 trigram 支持子串匹配对中文比较友好。FTS5 的经典用法是创建一个虚拟表虚拟表内部维护倒排索引。写入数据时我们同时维护业务表和 FTS 索引查询时直接用 MATCH 语法。3.3 基于 external content 表的设计这是 FTS5 的推荐用法FTS 虚拟表只做索引不存真实内容真实内容仍然存在普通表里。好处是不需要冗余存储大段文本且业务表增删改时用触发器同步索引。设计如下memories主表存记忆内容、类型、标签、时间等字段。memories_ftsFTS5 虚拟表通过contentmemories关联主表。三个触发器插入、更新、删除时自动同步 FTS 索引。这样上层代码不用关心索引同步细节只要操作主表FTS 索引自动保持一致。4. 完整实战从零实现一个 MCP Memory Server接下来是本文的核心。我们分几步完成初始化数据库、实现指纹函数、封装 MCP 工具、编写本地测试客户端验证。4.1 初始化数据库在mcp_memory_server.py中先定义数据库路径和初始化函数。数据库路径建议通过环境变量MEMORY_DB_PATH指定这样客户端配置时可以灵活调整。import os DB_PATH os.environ.get(MEMORY_DB_PATH, os.path.join(os.path.dirname(__file__), memory.db)) def init_db(): 初始化数据库表结构和 FTS5 索引。 conn sqlite3.connect(DB_PATH) conn.executescript( CREATE TABLE IF NOT EXISTS memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, memo_id TEXT NOT NULL UNIQUE, okf TEXT NOT NULL UNIQUE, content TEXT NOT NULL, kind TEXT NOT NULL DEFAULT note, tags TEXT NOT NULL DEFAULT , source TEXT NOT NULL DEFAULT manual, created_at TEXT NOT NULL DEFAULT (datetime(now)), updated_at TEXT NOT NULL DEFAULT (datetime(now)) ); -- FTS5 外部内容表tokenizer 使用 trigram支持中文子串匹配 CREATE VIRTUAL TABLE IF NOT EXISTS memories_fts USING fts5( content, kind, tags, contentmemories, content_rowidid, tokenizetrigram ); -- 插入触发器向 FTS 表写入索引 CREATE TRIGGER IF NOT EXISTS memories_ai AFTER INSERT ON memories BEGIN INSERT INTO memories_fts(rowid, content, kind, tags) VALUES (new.id, new.content, new.kind, new.tags); END; -- 删除触发器删除 FTS 索引中的对应行 CREATE TRIGGER IF NOT EXISTS memories_ad AFTER DELETE ON memories BEGIN INSERT INTO memories_fts(memories_fts, rowid, content, kind, tags) VALUES (delete, old.id, old.content, old.kind, old.tags); END; -- 更新触发器先删旧索引再插入新索引 CREATE TRIGGER IF NOT EXISTS memories_au AFTER UPDATE ON memories BEGIN INSERT INTO memories_fts(memories_fts, rowid, content, kind, tags) VALUES (delete, old.id, old.content, old.kind, old.tags); INSERT INTO memories_fts(rowid, content, kind, tags) VALUES (new.id, new.content, new.kind, new.tags); END; ) conn.commit() conn.close()这里的重点是memo_id是对外暴露的记忆 ID使用 UUID 字符串方便客户端引用。okf是内容指纹做唯一去重。memories_fts使用contentmemories和content_rowididFTS 索引直接关联主表的 rowid。触发器保证索引同步UPDATE触发器先delete再insert这是 FTS5 external content 表的官方推荐写法。再写一个获取数据库连接的辅助函数def get_conn() - sqlite3.Connection: conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row conn.execute(PRAGMA journal_modeWAL;) conn.execute(PRAGMA busy_timeout5000;) return conn启用 WAL 模式之后SQLite 支持读写并发适合 MCP Server 多会话场景。busy_timeout可以避免并发写入时直接报 “database is locked”。4.2 实现 MCP Server 入口使用 FastMCP 创建 Serverimport uuid from mcp.server.fastmcp import FastMCP mcp FastMCP(mcp-memory)4.3 实现记忆写入工具rememberremember工具接受内容、类型、标签、来源等参数。逻辑如下基于 content 和 kind 计算 OKF 指纹。查询数据库是否已存在该指纹。新增或更新记录。返回操作结果。mcp.tool() def remember( content: str, kind: str note, tags: str , source: str manual, ) - dict: 记录一条新记忆如果内容指纹相同则更新已有记录。 Args: content: 记忆内容例如“用户喜欢用 Python 写后端”。 kind: 记忆类型如 preference、fact、decision、note。 tags: 逗号分隔的标签例如“python, fastapi”。 source: 记忆来源例如 manual、claude、codex。 content content.strip() if not content: return {status: error, message: content cannot be empty} fp okf_fingerprint({content: content, kind: kind}) conn get_conn() existing conn.execute( SELECT memo_id FROM memories WHERE okf ?, (fp,) ).fetchone() if existing: conn.execute( UPDATE memories SET tags ?, source ?, updated_at datetime(now) WHERE okf ? , (tags, source, fp), ) conn.commit() conn.close() return { status: updated, memo_id: existing[memo_id], okf: fp, } memo_id str(uuid.uuid4()) conn.execute( INSERT INTO memories (memo_id, okf, content, kind, tags, source) VALUES (?, ?, ?, ?, ?, ?) , (memo_id, fp, content, kind, tags, source), ) conn.commit() conn.close() return { status: created, memo_id: memo_id, okf: fp, }这里的设计细节是指纹只对content kind计算tags 变化不算新记忆。这样做的好处是“内容一致但标签补全”时只是更新标签不会产生重复记忆。如果你希望 tags 变化也算新记忆可以把 tags 加进指纹对象按业务需要调整。4.4 实现记忆检索工具recallrecall是记忆系统最核心的读取接口。它通过 FTS5 全文检索实现对输入 query 做清洗拆成多个有效词。对长度 3 的词走 FTS5 MATCH词之间用 OR 连接保证召回率。如果 query 过短或为空退化为 LIKE 查询或直接返回最新记录。按 BM25 相关度升序排序FTS5 中 BM25 分数越低越相关。import re def build_match_query(query: str) - str: 把用户输入拆成 token并用 OR 连接成 FTS5 查询表达式。 trigram tokenizer 要求查询词至少 3 个字符所以这里过滤掉短词。 terms [] for token in re.split(r\s, query.strip()): token token.strip().strip() if len(token) 3: terms.append(f{token.replace(chr(34), chr(34) * 2)}) return OR .join(terms) mcp.tool() def recall(query: str, limit: int 10) - list[dict]: 根据关键词检索历史记忆按相关度排序。 Args: query: 关键词例如“用户喜欢什么编程语言”。 limit: 返回条数默认 10。 query query.strip() limit max(1, min(int(limit), 50)) conn get_conn() if len(query) 3: match_expr build_match_query(query) if match_expr: rows conn.execute( SELECT m.memo_id, m.content, m.kind, m.tags, m.source, m.created_at, m.updated_at, bm25(memories_fts) AS score FROM memories_fts JOIN memories m ON m.id memories_fts.rowid WHERE memories_fts MATCH ? ORDER BY score ASC LIMIT ? , (match_expr, limit), ).fetchall() conn.close() return [dict(row) for row in rows] # 退化为 LIKE 查询适合不足 3 个字符或空查询 like f%{query}% rows conn.execute( SELECT memo_id, content, kind, tags, source, created_at, updated_at, 0 AS score FROM memories WHERE content LIKE ? OR tags LIKE ? ORDER BY updated_at DESC LIMIT ? , (like, like, limit), ).fetchall() conn.close() return [dict(row) for row in rows]说明几点bm25(memories_fts)返回的分数越低代表相关度越高因此ORDER BY score ASC。为什么用 OR 而不是 AND记忆检索场景我们希望更“宽容”只要记忆包含任意一个关键词就应该召回再由 BM25 排序决定谁更靠前。如果用 AND稍微长一点的查询就很容易查不到结果。如果 query 很短比如 12 个字符trigram tokenizer 无法匹配我们直接退回 LIKE。4.5 实现删除与统计工具删除操作在生产环境必须谨慎。下面的forget工具先根据 memo_id 查存在性再执行删除实际应用中建议增加二次确认逻辑比如要求额外传入confirmTrue。mcp.tool() def forget(memo_id: str, confirm: bool False) - dict: 删除一条记忆。必须显式传入 confirmTrue 才会真正执行删除。 Args: memo_id: 记忆 ID由 remember 返回。 confirm: 是否确认删除必须为 True。 if not confirm: return {status: cancelled, message: 请传入 confirmTrue 确认删除} conn get_conn() existing conn.execute( SELECT memo_id FROM memories WHERE memo_id ?, (memo_id,) ).fetchone() if not existing: conn.close() return {status: not_found, memo_id: memo_id} conn.execute(DELETE FROM memories WHERE memo_id ?, (memo_id,)) conn.commit() conn.close() return {status: deleted, memo_id: memo_id}再加一个简单的统计工具方便后续做记忆质量分析和清理mcp.tool() def memory_stats() - dict: 返回记忆库的统计信息。 conn get_conn() total conn.execute(SELECT COUNT(*) AS total FROM memories).fetchone()[total] kinds conn.execute( SELECT kind, COUNT(*) AS cnt FROM memories GROUP BY kind ORDER BY cnt DESC ).fetchall() conn.close() return { total: total, by_kind: [dict(row) for row in kinds], }4.6 实现 Resources 和 Prompts除了 ToolsMCP Server 还可以暴露 Resources 和 Prompts让模型直接读取或按模板执行。mcp.resource(memory://recent) def recent_memories() - str: 返回最近 10 条记忆供模型直接读取。 conn get_conn() rows conn.execute( SELECT memo_id, content, kind, tags, updated_at FROM memories ORDER BY updated_at DESC LIMIT 10 ).fetchall() conn.close() lines [f- [{row[kind]}] {row[content]} (tags: {row[tags]}) for row in rows] return \n.join(lines) if lines else 暂无记忆。 mcp.prompt() def memory_retrieval(question: str) - str: 当用户提问时先检索历史记忆再回答。 return f请先调用 recall 工具检索与「{question}」相关的历史记忆再结合记忆内容回答用户。如果检索结果为空请直接说明没有找到相关记忆。4.7 启动入口最后加上启动入口if __name__ __main__: init_db() mcp.run(transportstdio)启动脚本完整文件如下路径mcp_memory_server.pyimport hashlib import json import os import re import sqlite3 import uuid from mcp.server.fastmcp import FastMCP DB_PATH os.environ.get(MEMORY_DB_PATH, os.path.join(os.path.dirname(__file__), memory.db)) mcp FastMCP(mcp-memory) def okf_fingerprint(obj: object) - str: canonical json.dumps( obj, sort_keysTrue, separators(,, :), ensure_asciiFalse, ) return hashlib.blake2b(canonical.encode(utf-8), digest_size16).hexdigest() def get_conn() - sqlite3.Connection: conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row conn.execute(PRAGMA journal_modeWAL;) conn.execute(PRAGMA busy_timeout5000;) return conn def init_db(): conn sqlite3.connect(DB_PATH) conn.executescript( CREATE TABLE IF NOT EXISTS memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, memo_id TEXT NOT NULL UNIQUE, okf TEXT NOT NULL UNIQUE, content TEXT NOT NULL, kind TEXT NOT NULL DEFAULT note, tags TEXT NOT NULL DEFAULT , source TEXT NOT NULL DEFAULT manual, created_at TEXT NOT NULL DEFAULT (datetime(now)), updated_at TEXT NOT NULL DEFAULT (datetime(now)) ); CREATE VIRTUAL TABLE IF NOT EXISTS memories_fts USING fts5( content, kind, tags, contentmemories, content_rowidid, tokenizetrigram ); CREATE TRIGGER IF NOT EXISTS memories_ai AFTER INSERT ON memories BEGIN INSERT INTO memories_fts(rowid, content, kind, tags) VALUES (new.id, new.content, new.kind, new.tags); END; CREATE TRIGGER IF NOT EXISTS memories_ad AFTER DELETE ON memories BEGIN INSERT INTO memories_fts(memories_fts, rowid, content, kind, tags) VALUES (delete, old.id, old.content, old.kind, old.tags); END; CREATE TRIGGER IF NOT EXISTS memories_au AFTER UPDATE ON memories BEGIN INSERT INTO memories_fts(memories_fts, rowid, content, kind, tags) VALUES (delete, old.id, old.content, old.kind, old.tags); INSERT INTO memories_fts(rowid, content, kind, tags) VALUES (new.id, new.content, new.kind, new.tags); END; ) conn.commit() conn.close() def build_match_query(query: str) - str: terms [] for token in re.split(r\s, query.strip()): token token.strip().strip() if len(token) 3: terms.append(f{token.replace(chr(34), chr(34) * 2)}) return OR .join(terms) mcp.tool() def remember(content: str, kind: str note, tags: str , source: str manual) - dict: content content.strip() if not content: return {status: error, message: content cannot be empty} fp okf_fingerprint({content: content, kind: kind}) conn get_conn() existing conn.execute( SELECT memo_id FROM memories WHERE okf ?, (fp,) ).fetchone() if existing: conn.execute( UPDATE memories SET tags ?, source ?, updated_at datetime(now) WHERE okf ? , (tags, source, fp), ) conn.commit() conn.close() return {status: updated, memo_id: existing[memo_id], okf: fp} memo_id str(uuid.uuid4()) conn.execute( INSERT INTO memories (memo_id, okf, content, kind, tags, source) VALUES (?, ?, ?, ?, ?, ?) , (memo_id, fp, content, kind, tags, source), ) conn.commit() conn.close() return {status: created, memo_id: memo_id, okf: fp} mcp.tool() def recall(query: str, limit: int 10) - list[dict]: query query.strip() limit max(1, min(int(limit), 50)) conn get_conn() if len(query) 3: match_expr build_match_query(query) if match_expr: rows conn.execute( SELECT m.memo_id, m.content, m.kind, m.tags, m.source, m.created_at, m.updated_at, bm25(memories_fts) AS score FROM memories_fts JOIN memories m ON m.id memories_fts.rowid WHERE memories_fts MATCH ? ORDER BY score ASC LIMIT ? , (match_expr, limit), ).fetchall() conn.close() return [dict(row) for row in rows] like f%{query}% rows conn.execute( SELECT memo_id, content, kind, tags, source, created_at, updated_at, 0 AS score FROM memories WHERE content LIKE ? OR tags LIKE ? ORDER BY updated_at DESC LIMIT ? , (like, like, limit), ).fetchall() conn.close() return [dict(row) for row in rows] mcp.tool() def forget(memo_id: str, confirm: bool False) - dict: if not confirm: return {status: cancelled, message: 请传入 confirmTrue 确认删除} conn get_conn() existing conn.execute( SELECT memo_id FROM memories WHERE memo_id ?, (memo_id,) ).fetchone() if not existing: conn.close() return {status: not_found, memo_id: memo_id} conn.execute(DELETE FROM memories WHERE memo_id ?, (memo_id,)) conn.commit() conn.close() return {status: deleted, memo_id: memo_id} mcp.tool() def memory_stats() - dict: conn get_conn() total conn.execute(SELECT COUNT(*) AS total FROM memories).fetchone()[total] kinds conn.execute( SELECT kind, COUNT(*) AS cnt FROM memories GROUP BY kind ORDER BY cnt DESC ).fetchall() conn.close() return {total: total, by_kind: [dict(row) for row in kinds]} mcp.resource(memory://recent) def recent_memories() - str: conn get_conn() rows conn.execute( SELECT memo_id, content, kind, tags, updated_at FROM memories ORDER BY updated_at DESC LIMIT 10 ).fetchall() conn.close() lines [f- [{row[kind]}] {row[content]} (tags: {row[tags]}) for row in rows] return \n.join(lines) if lines else 暂无记忆。 mcp.prompt() def memory_retrieval(question: str) - str: return f请先调用 recall 工具检索与「{question}」相关的历史记忆再结合记忆内容回答用户。如果检索结果为空请直接说明没有找到相关记忆。 if __name__ __main__: init_db() mcp.run(transportstdio)4.8 编写测试客户端验证由于 MCP Server 是基于 stdio 传输的直接在命令行里不好模拟双向通信。我建议写一个简单的测试客户端用 MCP Python SDK 连接并调用工具。# 文件路径test_client.py import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): server_params StdioServerParameters( commandpython, args[mcp_memory_server.py], env{MEMORY_DB_PATH: memory.db}, ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() print(可用工具, [tool.name for tool in tools.tools]) res await session.call_tool( remember, arguments{ content: 用户喜欢用 Python 和 FastAPI 写后端服务, kind: preference, tags: python, fastapi, source: test, }, ) print(remember 结果, res.content) res await session.call_tool( recall, arguments{query: 用户喜欢什么语言, limit: 5}, ) print(recall 结果, res.content) if __name__ __main__: asyncio.run(main())运行方式python test_client.py预期会输出类似下面的内容可用工具 [remember, recall, forget, memory_stats] remember 结果 [TextContent(typetext, text{status: created, memo_id: xxxx, okf: xxxx})] recall 结果 [TextContent(typetext, text[{memo_id: xxxx, content: 用户喜欢用 Python 和 FastAPI 写后端服务, ...}])]到这里一个功能完整的 MCP Memory Server 已经跑通了。但你可能注意到一点我们目前还没有真正让 Claude、Codex、Cline 这些客户端使用它。下面就来解决接入问题。5. 将 MCP Server 接入客户端有了 MCP Server下一步就是把服务注册到 AI 客户端里。不同客户端的配置方式略有差异但核心思路一致指定command、args、env让客户端通过 stdio 拉起我们的 Server。5.1 接入 Claude DesktopClaude Desktop 的配置文件路径macOS~/Library/Application Support/Claude/claude_desktop_config.jsonWindows%APPDATA%\Claude\claude_desktop_config.json在配置文件的mcpServers中新增一项{ mcpServers: { mcp-memory: { command: python, args: [ /绝对路径/mcp_memory_server.py ], env: { MEMORY_DB_PATH: /绝对路径/memory.db } } } }保存后重启 Claude Desktop然后在对话中输入“帮我把这条记忆记下来用户喜欢用 Python”模型就会调用remember工具。新会话里问“我平时喜欢用什么语言”模型会调用recall工具从数据库里把记忆找回来。这里有一个很重要的坑command里的python必须能直接在客户端环境里启动。如果你是用虚拟环境安装的 mcp 包建议把command改成虚拟环境里的绝对路径例如/path/to/.venv/bin/pythonWindows 下是.venv\Scripts\python.exe否则客户端找不到依赖包。5.2 接入 Codex 与 ClineCodex 是 OpenAI 的命令行编程工具支持通过--mcp-config指定 MCP 配置文件。不同版本对配置文件的格式要求略有差异建议先查询你当前版本的支持情况。一般形式是codex --mcp-config mcp.json其中mcp.json内容类似{ mcpServers: { mcp-memory: { command: python, args: [/绝对路径/mcp_memory_server.py], env: { MEMORY_DB_PATH: /绝对路径/memory.db } } } }Cline 是 VS Code 里常用的 AI 编程插件它的 MCP Server 配置界面在插件设置中。添加方式打开 Cline 设置。找到 MCP Servers 配置。选择类型为stdio。Command 填python。Args 填[/绝对路径/mcp_memory_server.py]。Environment 里设置MEMORY_DB_PATH。注意如果你在 Codex、Cline 这类工具里发现“工具注册不上”常见原因有三个一是脚本启动时报错导致注册失败二是 Python 环境不对三是 Server 初始化耗时过长客户端超时。建议先用下面的 Inspector 方式调试。5.3 用 MCP Inspector 快速调试MCP Python SDK 提供了 Inspector 调试工具可以可视化查看 Server 暴露了哪些工具、参数是什么。运行方式mcp dev mcp_memory_server.py如果你的 SDK 版本支持该命令它会启动一个本地 Web 面板你可以在面板里看到工具列表手动调用并观察返回结果排查问题比直接接客户端快很多。6. 常见问题与排查思路问题现象常见原因解决思路启动时报no such module: fts5Python 自带的 sqlite3 没有启用 FTS5升级 Python 版本或安装 pysqlite3 并替换 sqlite3 模块客户端显示工具注册失败脚本启动
