MCP Token超限问题解决方案
🌐 什么是MCP？
Model Context Protocol (MCP) 是一个开放标准，允许AI模型安全地连接到外部数据源和工具。MCP服务器可以提供丰富的工具集合，如GitHub仓库查询、数据库操作、文档检索等。
然而，MCP工具经常返回大量数据（如完整的GitHub文档、API响应等），直接传递给LLM会导致token超限错误。
🔥 MCP场景下的Token超限问题
典型场景
- GitHub文档查询 - MCP工具返回完整的README、API文档
- 代码库分析 - 返回大量源代码内容和注释
- 数据库查询 - 返回成千上万行查询结果
- API调用 - 返回复杂的JSON响应数据
错误示例
Token超限错误
Error code: 400 - {'error': {'message': "This model's maximum context length is 16385 tokens. However, your messages resulted in 72131 tokens (72090 in the messages, 41 in the functions). Please reduce the length of the messages or functions.",'type': 'invalid_request_error','code': 'context_length_exceeded'}
}1 用户向Agent发送查询请求
2 Agent选择合适的MCP工具
3 MCP工具返回大量数据（几万到几十万字符）
4 ❌ Agent尝试将完整数据传递给LLM → Token超限
5 ✅ 使用拦截器摘要数据 → 成功生成回答
🏆 两种生产级解决方案
方案一：LangChain _aperform_agent_action 拦截器
适用场景：传统LangChain AgentExecutor + MCP工具集成
核心思路：继承AgentExecutor，重写_aperform_agent_action方法，在工具执行后立即拦截和处理返回数据，返回正确的AgentStep对象。
技术优势：工具级别精确控制，保持Agent完整功能，支持流式输出。
方案一：MCP Agent拦截器核心实现
from langchain_core.agents import AgentStep, AgentAction
from langchain.agents import AgentExecutor
from langchain_openai import ChatOpenAIclass MCPAgentInterceptor(AgentExecutor):"""MCP专用Agent拦截器"""SUMMARY_THRESHOLD: ClassVar[int] = 1500async def _aperform_agent_action(self, name_to_tool_map, color_mapping, agent_action, run_manager=None) -> AgentStep:"""拦截MCP工具输出并进行智能摘要"""# 执行MCP工具agent_step = await super()._aperform_agent_action(name_to_tool_map, color_mapping, agent_action, run_manager)# 检查是否需要摘要if len(agent_step.observation) > self.SUMMARY_THRESHOLD:user_query = str(agent_action.tool_input)summarized = await self._summarize_mcp_output(agent_step.observation, user_query, agent_action.tool)# 🔥 关键：返回AgentStep对象return AgentStep(action=agent_step.action,observation=summarized)return agent_stepasync def _summarize_mcp_output(self, mcp_output, user_query, tool_name):"""智能摘要MCP返回数据"""summarizer = ChatOpenAI(model="gpt-4o", temperature=0)prompt = f"""对MCP工具输出进行摘要:
用户问题: {user_query}
工具名称: {tool_name}
原始输出: {mcp_output}要求: 保留核心信息，控制在800字符以内"""response = await summarizer.ainvoke([{"role": "user", "content": prompt}])return f"[{tool_name} MCP摘要] {response.content}"# MCP Session使用
async with sse_client(mcp_url) as (read, write):async with ClientSession(read, write) as session:await session.initialize()mcp_tools = await load_mcp_tools(session)# 使用MCP拦截器agent = MCPAgentInterceptor(agent=base_agent, tools=mcp_tools)result = await agent.ainvoke({"input": query})方案二：LangGraph pre_model_hook 消息修剪
适用场景：LangGraph生态系统 + MCP工具集成
核心思路：使用LangGraph原生的pre_model_hook参数，在LLM调用前自动修剪消息历史，防止token超限。
技术优势：框架原生支持，消息级别处理，实现简洁优雅。
方案二：LangGraph pre_model_hook核心实现
from langchain_core.messages import RemoveMessage
from langchain_core.messages.utils import trim_messages, count_tokens_approximately
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.prebuilt import create_react_agentdef pre_model_hook(state):"""MCP场景下的消息修剪处理"""messages = state["messages"]token_count = count_tokens_approximately(messages)if token_count > 4000: # MCP数据通常很大，设置较低阈值processed_messages = []# 智能压缩MCP工具返回的大数据for msg in messages:if hasattr(msg, 'content') and len(str(msg.content)) > 5000:if hasattr(msg, 'type') and msg.type == 'tool':# 压缩MCP工具数据compressed_content = compress_mcp_data(str(msg.content))compressed_msg = msg.__class__(content=compressed_content,**{k: v for k, v in msg.__dict__.items() if k != 'content'})processed_messages.append(compressed_msg)else:processed_messages.append(msg)else:processed_messages.append(msg)# 进一步修剪消息历史if count_tokens_approximately(processed_messages) > 8000:trimmed_messages = trim_messages(processed_messages,strategy="last",max_tokens=6000,include_system=True)processed_messages = trimmed_messages if trimmed_messages else processed_messages[-2:]return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES)] + processed_messages}return statedef compress_mcp_data(content: str) -> str:"""压缩MCP工具输出数据"""try:import jsondata = json.loads(content)# 保留关键字段，限制长度summary = {"type": "mcp_compressed", "original_length": len(content)}for key in ["title", "description", "content", "data"]:if key in data:value = str(data[key])summary[key] = value[:200] + "..." if len(value) > 200 else valuereturn json.dumps(summary, ensure_ascii=False)except:return content[:800] + f"...[MCP数据已压缩，原长{len(content)}字符]"# 创建LangGraph Agent
agent = create_react_agent(model=model,tools=mcp_tools,pre_model_hook=pre_model_hook
)📊 方案对比分析
| 特性维度 | LangChain _aperform_agent_action | LangGraph pre_model_hook |
|---|---|---|
| 适用框架 | LangChain AgentExecutor + MCP | LangGraph + MCP |
| 处理级别 | 工具输出级别 | 消息历史级别 |
| 实现复杂度 | 中等 - 需重写方法 | 简单 - 原生参数 |
| 控制精度 | 高精度 - 针对单个工具 | 中等 - 全局消息处理 |
| MCP会话管理 | ✅ 完全兼容session生命周期 | ✅ 完全兼容session生命周期 |
| 流式输出支持 | ✅ ainvoke, astream, astream_events | ✅ LangGraph原生流式支持 |
| Agent功能保持 | ✅ 完整保持（正确返回AgentStep） | ✅ 框架层面保证 |
| 错误处理 | ✅ 工具级别异常处理 | ✅ 消息级别降级策略 |
🏆 总结与建议
🎯 方案选择建议
新项目推荐：优先选择 LangGraph + pre_model_hook 方案
- ✅ 原生框架支持，代码简洁
- ✅ 更好的长期维护性
- ✅ 与LangGraph生态系统深度集成
现有项目迁移：使用 LangChain + _aperform_agent_action 方案
- ✅ 无需改变现有架构
- ✅ 精确的工具级别控制
- ✅ 与现有AgentExecutor完美兼容
🚀 技术价值
通过本文的两种解决方案，可以彻底解决MCP场景下的token超限问题，同时：
- 📈 大幅提升系统稳定性 - 避免99%的token超限错误
- ⚡ 优化响应性能 - 压缩比可达数百倍
- 🔧 保持Agent完整功能 - 工具选择、重试、流式输出全支持
- 🌐 真实场景验证 - 经过真实MCP服务器大数据测试