1. 为什么你的 AI Agent 聊三句就“失忆”
如果你正在用 Python 写 AI Agent,大概率遇到过这种尴尬:第一轮问“帮我查一下项目里 user_service.py 的登录逻辑”,Agent 答得头头是道;第二轮追问“那它的 token 过期时间在哪配置”,它却像换了个人,反问你“哪个文件”。这不是模型笨,而是多轮上下文没有在 MCP Servers 与 AI Agent 之间正确传递。
MCP(Model Context Protocol)解决的是“AI 怎么标准化地拿到外部数据和工具”的问题,而连续对话解决的是“这些数据怎么在多轮里不丢”的问题。两者叠在一起,才是能真正干活的 Agent。本文聚焦 Python 环境下 MCP Servers 与 AI Agent 的连续对话集成,面向需要多轮上下文保持的开发者,交付可复制的 MCP Server 配置骨架(含 settings.json / config.toml 示例)、TaoToken 统一 Key/API 通道接入步骤,以及连续对话的验证动作与预期结果。读完你能自己搭一个“记得住上文、调得动工具”的 Python Agent。
我试过把 MCP 的 session 和对话历史分开管理,结果第二轮工具调用直接报session not found,后来才理清:MCP 管的是工具会话,对话历史得由 Agent 侧自己维护并每轮注入。下面按这个思路一步步来。
2. TaoToken 前置:统一 Key 与 API 通道
在写 MCP Server 之前,先把模型通道固定下来。MCP Server 本身不产生模型能力,它负责暴露工具和数据;真正做推理的是背后的模型。用 TaoToken 的好处是:一个 Key 走统一 API 通道,Python 侧不用为每个模型改 base_url 和鉴权逻辑,MCP Server 里读环境变量就行。
官网入口:https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=
API 地址(不带 UTM):https://taotoken.net/api
你需要先拿到 API Key,再去控制台确认通道可用。相关 deep link:
- 模型对话(验证模型是否通):https://taotoken.net/model-chat?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite
- API Keys 管理:https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite
- 接入文档:https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite
- 控制台:https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite
注意:Key 只放环境变量,别写进 settings.json 提交到 Git。MCP Server 配置里用
${TAOTOKEN_API_KEY}这种占位符引用。
如果你后续要做长期编码或 Agent 常驻任务,可以看 Coding Plan:https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite
3. 可复制配置:MCP Server 骨架与 settings.json
先建目录结构,保持 MCP Server 和 Agent 分离:
mcp-agent-demo/ ├── mcp_server/ │ ├── server.py │ └── config.toml ├── agent/ │ ├── agent.py │ └── settings.json └── .env.env里放:
TAOTOKEN_API_KEY=你的Key TAOTOKEN_BASE_URL=https://taotoken.net/api3.1 MCP Server 的 config.toml
MCP Server 需要声明自己暴露哪些工具、连哪个模型通道。用 TOML 写配置,Python 侧用tomllib(3.11+)或tomli读:
[server] name = "demo-mcp-server" version = "0.1.0" transport = "stdio" [model] provider = "taotoken" base_url = "https://taotoken.net/api" api_key_env = "TAOTOKEN_API_KEY" default_model = "claude-3-5-sonnet" [tools.file_reader] enabled = true root = "./workspace" max_bytes = 200000 [tools.shell] enabled = falsetransport = "stdio"表示 MCP Server 通过标准输入输出和 Agent 通信,这是本地开发最省事的方式。api_key_env指向环境变量名,而不是明文 Key。
3.2 Agent 的 settings.json
Agent 侧要配置“连哪个 MCP Server”和“对话历史怎么存”:
{ "mcpServers": { "demo": { "command": "python", "args": ["mcp_server/server.py"], "env": { "TAOTOKEN_API_KEY": "${TAOTOKEN_API_KEY}", "TAOTOKEN_BASE_URL": "https://taotoken.net/api" } } }, "conversation": { "max_turns": 20, "persist_path": "./agent/history.jsonl", "inject_tool_results": true } }max_turns控制注入模型的历史轮数,persist_path把每轮对话落盘,inject_tool_results决定工具返回结果是否回填进下一轮上下文——这个开关是连续对话能不能“记住工具结果”的关键。
3.3 MCP Server 最小实现
import json import os import sys import tomllib from pathlib import Path CONFIG_PATH = Path(__file__).parent / "config.toml" def load_config(): with open(CONFIG_PATH, "rb") as f: return tomllib.load(f) def handle_tool_call(name, arguments): cfg = load_config() if name == "file_reader": root = Path(cfg["tools"]["file_reader"]["root"]).resolve() target = (root / arguments["path"]).resolve() if not str(target).startswith(str(root)): return {"error": "path out of root"} return {"content": target.read_text(encoding="utf-8")[:cfg["tools"]["file_reader"]["max_bytes"]]} return {"error": f"unknown tool: {name}"} def main(): for line in sys.stdin: line = line.strip() if not line: continue msg = json.loads(line) if msg.get("method") == "tools/call": result = handle_tool_call(msg["params"]["name"], msg["params"].get("arguments", {})) print(json.dumps({"id": msg["id"], "result": result}), flush=True) elif msg.get("method") == "tools/list": print(json.dumps({"id": msg["id"], "result": {"tools": ["file_reader"]}}), flush=True) if __name__ == "__main__": main()这段代码做了三件事:读 config.toml、按 JSON-RPC 风格处理tools/list和tools/call、用flush=True保证 stdio 实时通信。file_reader做了路径越界检查,避免读到 workspace 之外的文件。
4. 连续对话验证:请求与预期结果
配置写完,先验证模型通道,再验证 MCP 工具调用,最后验证多轮上下文。
4.1 验证模型通道
import os from openai import OpenAI client = OpenAI( api_key=os.environ["TAOTOKEN_API_KEY"], base_url=os.environ["TAOTOKEN_BASE_URL"], ) resp = client.chat.completions.create( model="claude-3-5-sonnet", messages=[{"role": "user", "content": "只回复两个字:通了"}], ) print(resp.choices[0].message.content)预期输出:通了。如果报 401,去 API Keys 页面确认 Key 状态;如果报模型不存在,去模型对话页面确认当前通道支持的模型名。
4.2 验证 MCP 工具调用
手动往 MCP Server 的 stdin 发一条:
echo '{"id":1,"method":"tools/call","params":{"name":"file_reader","arguments":{"path":"hello.txt"}}}' | python mcp_server/server.py预期输出类似:
{"id": 1, "result": {"content": "hello mcp"}}4.3 验证连续对话
这是核心。Agent 侧维护一个messages列表,每轮把工具结果以role: tool或role: user回填:
import json from openai import OpenAI client = OpenAI(api_key=os.environ["TAOTOKEN_API_KEY"], base_url=os.environ["TAOTOKEN_BASE_URL"]) history = [] def chat(user_input, tool_result=None): history.append({"role": "user", "content": user_input}) if tool_result: history.append({"role": "user", "content": f"[工具返回] {tool_result}"}) resp = client.chat.completions.create( model="claude-3-5-sonnet", messages=history[-20:], ) answer = resp.choices[0].message.content history.append({"role": "assistant", "content": answer}) return answer print(chat("记住一个数字:42")) print(chat("我刚才让你记的数字是多少?"))预期结果:第二轮回答里出现42。如果第二轮答不出来,说明历史没注入或max_turns太小。实测下来,把history[-20:]改成history全量注入,短对话场景立刻正常,但长对话要注意 token 成本。
5. 本篇常见错排查
报错一:ModuleNotFoundError: No module named 'tomllib'Python 3.10 及以下没有tomllib。装tomli并改导入:
pip install tomlitry: import tomllib except ModuleNotFoundError: import tomli as tomllib报错二:MCP Server 无响应,Agent 卡住多半是 stdio 没 flush。检查print(..., flush=True)是否每处都有。另外确认settings.json里command用的是绝对路径或正确的相对路径,args里的脚本路径相对于 Agent 启动目录。
报错三:第二轮对话丢失工具结果检查inject_tool_results是否为true,以及工具返回是否真的 append 进了history。常见坑是工具结果只打印没回填,模型自然看不到。
报错四:401 / 403Key 没读到或失效。在 Python 里打印os.environ.get("TAOTOKEN_API_KEY")前几位确认非空,再去 API Keys 页面核对。别把 Key 写死在 settings.json 里。
报错五:路径越界path out of rootfile_reader的root是相对路径,解析基准是 MCP Server 进程的工作目录。建议在 config.toml 里写绝对路径,或在 server.py 启动时os.chdir(Path(__file__).parent)。
报错六:对话历史无限增长导致超 tokenmax_turns设 20 只是截断注入,history本身还在涨。加一个落盘 + 定期裁剪逻辑,把history.jsonl按会话 ID 分文件,超过 N 轮就归档。
6. 把通道和工具固定下来,Agent 才稳
连续对话的本质不是模型记性好,而是你在每一轮把该给它的上下文都喂到位。MCP Servers 负责“工具和数据怎么标准化暴露”,TaoToken 负责“模型通道怎么统一鉴权”,Agent 侧负责“历史怎么维护和注入”。三者边界清晰,调试时才能快速定位是哪一层出问题。
如果你要长期跑编码类 Agent,建议把模型通道固定成 Coding Plan,省得每次换模型改配置:https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite
接入细节和参数说明以官方文档为准:https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite
先把file_reader跑通,再按同样骨架加shell、db_query等工具,每加一个就用 4.2 的单条 JSON 验证一次,别等全接完再排障。