1. 脚本式调用为什么在生产环境撑不住
很多人第一次写 AI Agent,都是从一段 Python 脚本开始的:读一个 URL,拼一段 prompt,调一次模型,把结果 print 出来。跑通那一刻确实很爽,但只要你想让它每天定时跑、处理几十个任务、出错能自己重试,这套东西很快就会散架。我见过太多项目卡在这个阶段,代码里全是try/except包着requests.get,prompt 硬编码在函数里,换个模型要改十几个文件。
问题的本质不是代码写得不好,而是脚本的线性执行模型和 Agent 的真实运行需求不匹配。脚本假设每一步都成功,但生产环境里网络会抖、页面会变、模型会返回不合规的 JSON。脚本假设流程是固定的,但真实任务往往需要根据中间结果决定下一步做什么——比如抓取失败要换策略,数据不完整要补采。
这就是 MCP 和 LangGraph 要解决的两件事。MCP(Model Context Protocol)把「工具」标准化,让抓取、存储、查询这些能力以统一接口暴露给模型,而不是散落在各个函数里。LangGraph 把「流程」显式化成状态机,节点是动作,边是流转逻辑,条件边负责在出错时把状态导向重试分支而不是直接崩溃。两者结合,脚本才真正升级成可运维的智能体。
这篇文章面向的是已经写过基础 LLM 调用、想往生产级 Agent 迈一步的开发者。我会给出可复制的 MCP 服务配置、LangGraph 的节点与边定义,以及本地跑通和断点续跑的验证动作。最后会说明怎么把 endpoint 和鉴权统一改到 TaoToken 管理,避免每个项目各存一份 Key。整套东西你跟着做就能跑起来,不需要先理解全部理论。
2. 用 TaoToken 统一管理 endpoint 与鉴权
在动手写 MCP 服务和 LangGraph 之前,先把「模型从哪来、Key 放哪」这件事定下来。生产级 Agent 最忌讳的就是 Key 散落在各个.env里,换一个模型要翻遍整个仓库。我的做法是把所有模型调用收敛到一个统一的 endpoint 和一份鉴权配置上。
TaoToken 在这里扮演的角色是统一的模型接入层。你不需要在代码里硬编码某家厂商的地址,而是把 Base URL 指向https://taotoken.net/api,用一把 Key 管理多个模型的调用。这样 MCP 服务里的抓取工具、LangGraph 里的推理节点,都走同一个出口,排查问题时只需要看一个地方。
具体操作上,先去控制台创建 API Key。打开 https://taotoken.net/api-keys ,登录后新建一个 Key,复制出来。这个 Key 就是后面所有配置里要填的鉴权凭证。注意不要把它提交到 Git,放在环境变量或本地配置文件里。
然后确认你要用的模型 ID。不同任务对模型的要求不一样:抓取后的结构化抽取需要稳定的 JSON 输出,适合用指令遵循强的模型;而生成调研总结这种偏写作的节点,可以换一个更擅长长文本的模型。模型 ID 可以在模型对话页面里试出来,地址是 https://taotoken.net/models ,输入 prompt 看返回是否符合预期,确认后再写进配置。
如果你打算长期跑编码类或 Agent 类任务,Coding Plan 会更划算,入口在 https://taotoken.net/coding-plan 。它适合那种每天都要调用、token 消耗稳定的场景,比按量付费更可控。接入文档在 https://taotoken.net/doc ,里面有各语言的调用示例,配置前扫一眼能省不少调试时间。
把这三样东西准备好:Base URL(https://taotoken.net/api)、API Key、Model ID。后面 MCP 服务和 LangGraph 节点都会引用它们。统一管理的价值在于,当你要换模型或轮换 Key 时,只改一处,整个 Agent 的所有节点同步生效,不会出现某个工具还在用旧 Key 导致 401 的情况。
3. 可复制的 MCP 服务与 LangGraph 配置
这一节是核心,给出能直接抄的配置。先看 MCP 服务端。MCP 的本质是把工具以标准协议暴露出去,让 Claude Desktop、Cursor 这类客户端能直接调用。我们用 Python 写一个最小的 MCP Server,暴露一个scrape_page工具。
先装依赖,用 uv 管理环境:
uv init agent-demo cd agent-demo uv add mcp langgraph langchain-openai pydantic httpx然后写 MCP 服务文件mcp_server.py:
import httpx from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent app = Server("agent-tools") @app.list_tools() async def list_tools(): return [ Tool( name="scrape_page", description="抓取指定 URL 的正文文本,用于竞品页面分析", inputSchema={ "type": "object", "properties": { "url": {"type": "string", "description": "目标页面地址"} }, "required": ["url"], }, ) ] @app.call_tool() async def call_tool(name: str, arguments: dict): if name == "scrape_page": url = arguments["url"] async with httpx.AsyncClient(timeout=15) as client: resp = await client.get(url) resp.raise_for_status() return [TextContent(type="text", text=resp.text[:8000])] raise ValueError(f"unknown tool: {name}") async def main(): async with stdio_server() as (read, write): await app.run(read, write, app.create_initialization_options()) if __name__ == "__main__": import asyncio asyncio.run(main())这个服务通过 stdio 通信,客户端启动它之后就能调用scrape_page。接下来是客户端侧的配置。以 Claude Desktop 为例,编辑它的claude_desktop_config.json,路径在 macOS 上是~/Library/Application Support/Claude/claude_desktop_config.json,Windows 上是%APPDATA%\Claude\claude_desktop_config.json:
{ "mcpServers": { "agent-tools": { "command": "uv", "args": ["run", "python", "mcp_server.py"], "cwd": "/absolute/path/to/agent-demo" } } }注意cwd必须写绝对路径,否则客户端找不到你的脚本。配置完重启 Claude Desktop,在对话框里输入「用 scrape_page 抓取 https://example.com」,如果工具被正确调用,说明 MCP 这一层通了。
然后是 LangGraph 部分。我们要定义一个带重试的状态机。状态里存 URL、抓取结果、重试次数和最终报告。节点有三个:scrape、extract、save。条件边负责在抓取失败且重试次数没超限时回到scrape。
from typing import TypedDict, Optional from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from pydantic import BaseModel, Field class AgentState(TypedDict): url: str raw: Optional[str] report: Optional[str] retries: int class Report(BaseModel): product_name: str = Field(description="产品名称") pricing: list[str] = Field(description="定价梯度") summary: str = Field(description="Markdown 总结") llm = ChatOpenAI( model="你的模型ID", base_url="https://taotoken.net/api", api_key="你的API Key", temperature=0, ) def scrape_node(state: AgentState): import httpx try: resp = httpx.get(state["url"], timeout=15) resp.raise_for_status() return {"raw": resp.text[:8000], "retries": state["retries"]} except Exception: return {"raw": None, "retries": state["retries"] + 1} def extract_node(state: AgentState): structured = llm.with_structured_output(Report) result = structured.invoke( f"从以下网页内容提取竞品信息:\n{state['raw']}" ) return {"report": result.model_dump_json(indent=2)} def save_node(state: AgentState): with open("report.json", "w", encoding="utf-8") as f: f.write(state["report"]) return {} def route_after_scrape(state: AgentState): if state["raw"] is None and state["retries"] < 3: return "retry" if state["raw"] is None: return "fail" return "ok" graph = StateGraph(AgentState) graph.add_node("scrape", scrape_node) graph.add_node("extract", extract_node) graph.add_node("save", save_node) graph.set_entry_point("scrape") graph.add_conditional_edges( "scrape", route_after_scrape, {"retry": "scrape", "ok": "extract", "fail": END}, ) graph.add_edge("extract", "save") graph.add_edge("save", END) app = graph.compile()这段配置里,base_url和api_key就是第 2 节准备好的 TaoToken 信息。with_structured_output配合 Pydantic 的Report模型,强制模型返回符合 schema 的 JSON,类型不对会自动触发校验错误,你可以在extract_node外面再包一层重试。条件边route_after_scrape是生产级的关键:抓取失败时不是抛异常,而是把状态导回scrape节点,最多重试 3 次。
4. 本地跑通与断点续跑验证
配置写完了,现在验证它真的能跑。先单独测 MCP 服务,用官方提供的调试方式,在终端里直接启动:
uv run python mcp_server.py如果进程挂起等待输入,说明 stdio 服务正常启动了。更直观的验证是回到 Claude Desktop,让它调用工具抓一个真实页面,看返回的文本是不是页面正文。这一步通了,说明工具注册没问题。
接着跑 LangGraph。写一个入口脚本run.py:
from mcp_server import app as graph_app initial = {"url": "https://example.com", "raw": None, "report": None, "retries": 0} result = graph_app.invoke(initial) print(result["report"])执行uv run python run.py,如果一切正常,你会看到终端打印出结构化的 JSON 报告,同时当前目录生成report.json。这就是一次完整的「抓取 → 抽取 → 保存」流程。
断点续跑是生产级 Agent 的刚需。LangGraph 支持 checkpoint,把状态持久化到 SQLite,进程中断后能从最后一个成功的节点继续。加上 checkpoint 只需要改两行:
from langgraph.checkpoint.sqlite import SqliteSaver memory = SqliteSaver.from_conn_string("checkpoints.db") app = graph.compile(checkpointer=memory)然后调用时传入config={"configurable": {"thread_id": "task-001"}}。这样即使你在extract节点执行到一半 Ctrl+C,重新运行同样的thread_id,它会从断点恢复,不会重复抓取已经成功的页面。验证方法是:第一次跑到extract时手动中断,再跑一次,观察日志里scrape节点是否被跳过。如果跳过了,说明 checkpoint 生效。
实测下来,这套组合最舒服的地方是排障。以前脚本报错只能看堆栈,现在每个节点的输入输出状态都存着,哪一步的raw是空的、哪一步的report校验失败,一目了然。你可以把checkpoints.db用 SQLite 客户端打开,直接看状态快照。
5. 常见报错与排查对照
跑这套东西,报错基本集中在几个地方。我按真实遇到的频率列出来,对照着查。
401 Unauthorized。这是最常见的,九成是 Key 或 Base URL 写错了。检查ChatOpenAI里的base_url是不是https://taotoken.net/api,注意结尾不要多加/v1或斜杠。api_key确认是从 https://taotoken.net/api-keys 复制的最新值,没有多余空格。如果 Key 刚轮换过,记得所有引用它的地方都要更新。
local proxy failed / connection refused。这个报错通常出现在 MCP 客户端启动服务时。原因是claude_desktop_config.json里的cwd路径不对,或者uv不在系统 PATH 里。解决办法是把command改成uv的绝对路径,用which uv查出来填进去。另外确认cwd指向的目录里确实有mcp_server.py。
reading 'choices' of undefined。这是模型返回体结构不符合预期导致的,多半是 Base URL 指错了地方,请求打到了不兼容的接口上。确认base_url是 TaoToken 的地址,并且model字段填的是平台上真实存在的模型 ID。如果模型 ID 拼错,有些网关会返回一个空对象,解析choices时就炸了。
OAuth / authentication failed。如果你用的是 Claude Code 或某些需要 OAuth 的客户端,报这个错说明鉴权流程没走完。检查客户端里的登录状态,重新授权一次。如果是 API Key 模式,确认没有同时配置 OAuth 和 Key 导致冲突。
ValidationError from Pydantic。这不是网络问题,是模型返回的 JSON 不符合Reportschema。常见原因是 prompt 里没强调字段要求,或者模型能力不够。解决办法是在extract_node里加一层重试,捕获ValidationError后把错误信息拼回 prompt 让模型自我修正。另外确认temperature=0,降低随机性。
断点续跑不生效。检查thread_id是否每次调用都一致。如果每次生成新的 UUID,checkpoint 就找不到历史状态。另外确认checkpoints.db文件有写权限,路径不要放在只读目录。
排查时有个通用技巧:把 LangGraph 的stream模式打开,用graph_app.stream(initial)代替invoke,能看到每个节点的实时输出,哪一步卡住立刻定位。
6. 把 Agent 接入你的日常工作流
到这里,MCP 服务、LangGraph 状态机、TaoToken 统一鉴权三件事都跑通了。接下来是把它用起来。最直接的接入方式是把 MCP 服务挂到你的编辑器里,Cursor 和 Claude Desktop 都支持。配置方式和第 3 节一样,改一下cwd指向你的项目目录即可。挂上之后,你在写代码时可以直接对话调用抓取工具,不用切终端。
对于需要定时跑的任务,把run.py包一层 cron 或 systemd timer。因为有了 checkpoint,即使某次执行中途机器重启,下次启动会自动续跑,不会丢任务。日志建议单独输出到文件,配合thread_id做追踪,出问题时能快速定位是哪个任务、哪个节点失败。
模型选择上,抽取节点用指令遵循强的模型保证 JSON 稳定,总结节点可以换成长文本能力更好的模型。切换时只改ChatOpenAI的model参数,Base URL 和 Key 不动,这就是统一 endpoint 的好处。如果你要跑多个 Agent 实例,每个实例用不同的thread_id隔离状态,共享同一份鉴权配置。
长期跑下来,token 消耗会趋于稳定,这时候 Coding Plan 比按量付费更省心,入口在 https://taotoken.net/coding-plan 。接入过程中遇到配置问题,文档在 https://taotoken.net/doc 有各语言的完整示例。需要新建或轮换 Key 时,去 https://taotoken.net/api-keys 操作。想先试模型效果再决定用哪个,模型对话页面 https://taotoken.net/models 可以直接输入 prompt 验证。
从脚本到智能体,核心变化不是代码量,而是思维方式:把「一次调用」变成「一个可恢复、可观测、可重试的状态机」。MCP 负责把工具标准化,LangGraph 负责把流程显式化,TaoToken 负责把模型接入统一化。三件事各司其职,你的 Agent 才算真正能上生产。