1. 为什么我要给 Agent 加一层 Redis 短期记忆做 Agent 的朋友大概率都遇到过这个场景用户上一轮说“帮我查下白露的习俗”下一轮问“那它一般吃什么”模型直接懵了——因为你的编排层根本没把上一轮的上下文带过去。要么你把整段历史塞进 prompttoken 爆炸要么你干脆不带体验稀碎。短期记忆Short-Term Memory就是解决这个断层的最小可用方案它不追求长期知识沉淀只负责在会话窗口内把“刚刚说过什么”稳稳记住过期自动清掉。我这次选的技术组合是 fastmcp 2.0 Redis。fastmcp 2.0 是目前写 MCP Server 最省心的 Python 框架装饰器一贴就是工具stdio 和 HTTP 两种传输一行切换Redis 则天然适合做短期记忆——HASH 存 KV、LIST 存时间线、EXPIRE 做滚动过期三件套刚好覆盖“会话隔离 自动遗忘”这两个核心诉求。整套东西跑起来你的 Agent 就多了一个可以随时调用的“内存工具”。这篇文章会交付一条完整链路一个提供 5 个记忆工具的 MCP Serverserver.py、一个最小可跑的 Clientclient.py、Redis 键结构设计、启动命令以及一次端到端的读写验证。同时我会把模型调用的 endpoint 统一改到 TaoToken 的 API 通道这样你后面接 Claude Code、Cline 或者自研 Agent 时Key 和 Base URL 只需要维护一份。适合谁看正在搭 Agent 编排层、被上下文管理折磨、想用 MCP 标准化工具接口的开发者。下面直接上干货。2. 前置准备Redis 环境与 TaoToken 统一 Key 通道在写代码之前先把两个前置条件搞定一个是本地 Redis一个是模型调用的统一入口。很多人卡在第一步不是不会装 Redis而是没想清楚“记忆层”和“模型层”应该怎么解耦。我的建议是Redis 只管存模型调用走统一 API 通道两者互不干扰这样你换模型、换框架都不用动记忆逻辑。2.1 起一个本地 Redis30 秒最省事的方式是 Docker一条命令搞定docker run -p 6379:6379 -d redis:7-alpine如果你本机已经装了 Redis直接redis-server也行。验证一下连通性redis-cli ping # 返回 PONG 就说明 OK这里有个小坑redis:7-alpine默认没有密码本地开发够用但如果你后面要放到测试环境记得加--requirepass并在连接串里带上密码否则redis.asyncio会直接抛NOAUTH。2.2 把模型 endpoint 改到 TaoTokenMCP Server 本身不调模型但你的 Client 或 Agent 编排层一定会调。与其在每个项目里散落一堆 Key不如统一走 TaoToken 的 API 通道。它的官网入口是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_end API 基址是 https://taotoken.net/api 这个地址不加 UTM 参数直接填进配置即可。具体操作路径登录后进控制台在 API Keys 页面创建一个 Key然后你的模型调用配置就变成三件套——Base URL 填https://taotoken.net/apiAPI Key 填刚创建的那串Model ID 按你实际要用的模型填。这三样东西后面在 Client 里会用到先记下来。注意TaoToken 是统一的 API 通道不是让你绕过什么它的定位就是帮你把多个模型的调用收敛到一个 Key 上省得你到处配环境变量。控制台地址是 https://taotoken.net/console API Keys 页面在 https://taotoken.net/api-keys 文档在 https://taotoken.net/doc 。2.3 安装依赖pip install fastmcp redisfastmcp 2.0 的包名就是fastmcp别装成老的mcp包API 完全不一样。装完可以python -c import fastmcp; print(fastmcp.__version__)确认一下版本。3. 可复制配置server.py 的 5 个记忆工具与 Redis 键结构这一节是全文的核心我会把 server.py 完整贴出来并解释每个工具的设计意图。你可以直接复制到本地跑。3.1 Redis 键结构设计先明确键的约定这是整个记忆层的地基用途Key 格式Redis 类型说明KV 记忆mcp:mem:{session}:kvHASH存键值对如 topic白露时间线mcp:mem:{session}:tlLISTRPUSH 追加LTRIM 裁剪过期两个键都设 EXPIRE—每次读写滚动续期会话隔离靠{session}这一段实现不同 session 的键天然不冲突。TTL 滚动续期是短期记忆的灵魂——只要有人在用它就不过期一旦静默超过 TTLRedis 自动回收不需要你写清理任务。3.2 server.py 完整代码import os, json, time from typing import Optional, Literal import redis.asyncio as redis from fastmcp import FastMCP REDIS_URL os.getenv(REDIS_URL, redis://localhost:6379/0) TTL_DEFAULT int(os.getenv(MEM_TTL_S, 3600)) mcp FastMCP(redis-memory) _redis: Optional[redis.Redis] None def r() - redis.Redis: global _redis if _redis is None: _redis redis.from_url(REDIS_URL, decode_responsesTrue) return _redis def k_kv(session_id: str) - str: return fmcp:mem:{session_id}:kv def k_tl(session_id: str) - str: return fmcp:mem:{session_id}:tl async def _bump_ttl(session_id: str, ttl_s: int): await r().expire(k_kv(session_id), ttl_s) await r().expire(k_tl(session_id), ttl_s) mcp.tool async def mem_put(session_id: str, key: str, value: str, ttl_s: int TTL_DEFAULT) - str: 设置/覆盖短期记忆 KV会话隔离并滚动续期 TTL。 await r().hset(k_kv(session_id), key, value) await _bump_ttl(session_id, ttl_s) return ok mcp.tool async def mem_get(session_id: str, key: str, default: Optional[str] None, ttl_s: int TTL_DEFAULT) - str: 读取短期记忆 KV命中后会顺带续期 TTL。 val await r().hget(k_kv(session_id), key) if val is None: return default if default is not None else await _bump_ttl(session_id, ttl_s) return val mcp.tool async def mem_append( session_id: str, role: Literal[user, agent, note], text: str, max_items: int 200, ttl_s: int TTL_DEFAULT, ) - str: 追加对话片段到时间线并按 max_items 裁剪每次写入滚动续期 TTL。 item {ts: int(time.time()), role: role, text: text} pipe r().pipeline() pipe.rpush(k_tl(session_id), json.dumps(item, ensure_asciiFalse)) pipe.ltrim(k_tl(session_id), -max_items, -1) pipe.expire(k_tl(session_id), ttl_s) pipe.expire(k_kv(session_id), ttl_s) await pipe.execute() return ok mcp.tool async def mem_recent(session_id: str, n: int 20, ttl_s: int TTL_DEFAULT) - str: 返回最近 n 条时间线JSON 数组字符串读取也会续期 TTL。 vals await r().lrange(k_tl(session_id), -n, -1) await _bump_ttl(session_id, ttl_s) items [json.loads(v) for v in vals] return json.dumps(items, ensure_asciiFalse) mcp.tool async def mem_clear(session_id: str) - str: 删除该会话的 KV 与时间线不可恢复。 pipe r().pipeline() pipe.delete(k_kv(session_id)) pipe.delete(k_tl(session_id)) await pipe.execute() return ok if __name__ __main__: mcp.run(transporthttp, host127.0.0.1, port8010, path/mcp)几个设计点值得展开说。第一_bump_ttl对两个键都尝试续期即使某个键还不存在也不会报错EXPIRE对不存在的键返回 0静默忽略这保证了幂等。第二mem_append用 pipeline 把 RPUSH、LTRIM、EXPIRE 打包成一次往返避免网络抖动导致时间线写了一半。第三mem_recent用lrange(key, -n, -1)取尾部 N 条配合 LTRIM 的-max_items, -1时间线永远只保留最近的记录不会无限膨胀。3.3 启动命令export REDIS_URLredis://localhost:6379/0 export MEM_TTL_S3600 python server.py启动后你会看到 fastmcp 打印监听信息Server 就跑在http://127.0.0.1:8010/mcp。如果你要接 Claude Desktop 这类本地客户端把mcp.run()改成不带参数的 stdio 模式即可工具定义完全不用动。4. 验证请求client.py 端到端读写与成功结果Server 起来了接下来用一个最小 Client 验证整条链路。这一步很关键很多人写完 Server 就以为完事了结果 Client 一调就报错其实问题往往出在传输方式或参数格式上。4.1 client.py 完整代码import asyncio, json from fastmcp import Client async def main(): async with Client(http://127.0.0.1:8010/mcp) as c: sid demo-session-1 await c.call_tool(mem_put, {session_id: sid, key: topic, value: 白露}) topic await c.call_tool(mem_get, {session_id: sid, key: topic}) print(topic , topic.text) await c.call_tool(mem_append, {session_id: sid, role: user, text: 什么是白露}) await c.call_tool(mem_append, {session_id: sid, role: agent, text: 白露是二十四节气之一...}) recent await c.call_tool(mem_recent, {session_id: sid, n: 5}) print(recent , json.loads(recent.text)) asyncio.run(main())4.2 预期输出topic 白露 recent [{ts: 1730000000, role: user, text: 什么是白露}, {ts: 1730000001, role: agent, text: 白露是二十四节气之一...}]看到这两行就说明端到端通了KV 写入读回一致时间线按顺序追加并能取回最近 N 条。你可以再手动跑一次mem_clear然后mem_recent应该返回空数组验证清理逻辑。4.3 在 Agent 编排层怎么接在你的 Orchestrator 里每轮问答结束后追加两条记忆await mcp_client.call_tool(mem_append, { session_id: context_id, role: user, text: user_query }) await mcp_client.call_tool(mem_append, { session_id: context_id, role: agent, text: agent_answer })下一轮生成前把最近若干条拉进上下文res await mcp_client.call_tool(mem_recent, {session_id: context_id, n: 10}) mem_snippets json.loads(res.text)这里的context_id建议直接用你 A2A 层的会话 ID天然隔离。如果你用的是 TaoToken 的 Coding Plan 跑长期编码任务可以把session_id设成项目名这样跨多次对话的记忆也能续上。Coding Plan 的入口在 https://taotoken.net/coding-plan 适合需要长时间保持上下文的 Agent 场景。5. 本篇常见错排查401、local proxy failed 与 reading choices这一节我把实际踩过的坑列出来对照报错直接定位。报错一401 Unauthorized。这个通常不是 MCP Server 的问题而是你的模型调用层 Key 没配对。检查三件套Base URL 是不是https://taotoken.net/apiAPI Key 是不是从 https://taotoken.net/api-keys 复制完整注意别带空格Model ID 是不是你账号下有权限的模型。三者缺一不可尤其是 Base URL 末尾别多加/v1TaoToken 的路径已经内置了。报错二local proxy failed。这个报错一般出现在 Client 连 Server 的阶段说明http://127.0.0.1:8010/mcp这个地址连不上。先确认 Server 进程还活着再确认端口没被占用lsof -i:8010最后检查你是不是把 transport 写成了 stdio 却用 HTTP 地址去连。传输方式必须两端一致。报错三reading choices相关错误。这类报错通常出现在模型返回体解析阶段说明你拿到的响应不是预期的 chat completion 结构。常见原因是 Model ID 填错或者你把 embedding 模型当 chat 模型用了。回到控制台确认模型类型chat 类模型才能返回 choices 字段。报错四NOAUTH Authentication required。Redis 加了密码但连接串没带。改成redis://:yourpasswordlocalhost:6379/0即可注意密码前面那个冒号不能省。报错五工具调用返回空字符串。mem_get返回空说明这个 key 没写过或者已经过期被 Redis 回收了。先用redis-cli keys mcp:mem:*看看键还在不在再用ttl mcp:mem:demo-session-1:kv看剩余过期时间。如果 TTL 是 -2说明键已消失属于正常过期行为。排查顺序建议先确认 Redis 通redis-cli ping再确认 Server 通浏览器访问http://127.0.0.1:8010/mcp应该有响应最后确认模型通道通用 curl 打一次 TaoToken 的 API。三层逐层排除比盲目改代码快得多。6. 把记忆层接进你的 Agent从 demo 到可用跑通 demo 只是起点真正要用起来还得考虑几件事。第一是会话隔离的粒度我建议把tenant_id也拼进 key变成mcp:mem:{tenant}:{session}:kv这样多租户场景下不会串数据。第二是安全HTTP 传输默认没有鉴权生产环境要么在反代层加 token 校验要么用 fastmcp 的认证钩子同时对session_id做格式白名单防止有人构造恶意 key。第三是容量治理。max_items和ttl_s都要设上限别让某个会话把 Redis 撑爆。如果你对 token 预算敏感可以在mem_append入库前对 text 做截断或摘要比如超过 500 字符就只存前 500 字加省略号。第四是观测给每个工具加时延和命中率指标用 Prometheus 收集再打通 trace这样出问题能快速定位是工具慢还是 Redis 慢。最后说下模型通道的维护。既然 endpoint 已经统一到 TaoToken你后面换模型只需要改 Model IDBase URL 和 Key 都不用动。想先试试模型对话效果可以去 https://taotoken.net/model-chat 直接体验要接 Claude Code 这类编码工具参考 https://taotoken.net/claude-code-anthropic 的配置说明把 Base URL 和 Key 填进去就行。整套记忆层 统一通道的组合实测下来能让 Agent 的上下文管理清爽不少至少不用再为“上一轮说了啥”这种问题写一堆胶水代码了。