如何在 MCP Toolbox 智能体架构中实现工具级预处理与后处理中间件

如何在 MCP Toolbox 智能体架构中实现工具级预处理与后处理中间件【免费下载链接】mcp-toolboxMCP Toolbox for Databases is an open source MCP server for databases.项目地址: https://gitcode.com/GitHub_Trending/ge/mcp-toolbox如果你的智能体通过 MCP Toolbox 连接数据库工具需要在工具执行前拦截参数例如校验业务规则、过滤敏感输入或在工具返回后修改结果例如补充字段、统一格式本文给出项目文档中提供的实现路径在编排框架的钩子或中间件层接入预处理与后处理逻辑。文档明确指出这些能力通常属于编排框架LangChain、LangGraph、ADK 等而非 Toolbox SDK 本身Toolbox 工具的设计是配合这些框架能力来构建更健壮的智能体架构。开始前需要满足以下条件对应项目的本地快速开始已按本地快速开始文档完成 MCP Toolbox 部署并配置好一个基础智能体Toolbox 服务运行在http://127.0.0.1:5000并且存在名为my-toolset的工具集文档示例统一使用该地址与工具集名如果你的配置不同需要把示例中对应值替换为你自己的地址和工具集名已配置 Google GenAI 的 API KeyPython 示例通过CredentialStrategy.toolbox_identity()处理凭据JS 示例从环境变量GOOGLE_API_KEY读取GOOGLE_GENAI_API_KEY。先明确处理范围工具级、模型级与智能体级文档把处理逻辑划分为三类范围本文只展开工具级这也是对工具执行做细粒度控制时最相关的层级工具级本文重点包裹单个工具的执行拦截工具的原始输入参数和输出。适用于参数校验、输出格式化、针对敏感工具的隐私规则。模型级拦截每次 LLM 调用提示词与响应对所有收发文本全局生效适合全局 PII 脱敏或 token 统计。ADK 中对应before_model_callback/after_model_callbackJS ADK 对应beforeModelCallback/afterModelCallbackLangChain 对应wrap_model/wrapModelCall。智能体级包裹高层执行循环如一轮对话从用户输入到最终响应适合会话管理或端到端审计。ADK 中对应before_agent_callback/after_agent_callbackJS ADK 对应beforeAgentCallback/afterAgentCallback。按文档定义典型的工具级预处理包括输入清洗与脱敏mask PII、业务规则校验例如酒店住宿不超过 14 天、安全护栏检测提示注入后处理包括响应增强向工具输出注入额外数据、输出格式化把 JSON/XML 转成更利于模型理解的格式、合规审计把请求与结果写入审计日志。Python ADK用 before_tool_callback 和 after_tool_callback 拦截工具调用这是文档中给出的最短主路径ToolboxToolset与 ADK 的 pre/post 处理钩子配合在工具调用前后插入逻辑。依赖版本来自示例目录的 requirements.txtpip install google-adk[toolbox]1.28.1 google-genai2.3.0完整可运行的示例见 agent.py。下面是其中与预处理、后处理直接相关的部分一个业务规则校验回调update-hotel的住宿时长超过 14 天则拦截一个响应增强回调book-hotel成功时附加积分信息# Pre processing async def enfore_business_rules( tool: ToolboxTool, args: Dict[str, Any], tool_context: ToolContext ) - Optional[Dict[str, Any]]: Callback fired before a tool is executed. Enforces business logic: Max stay duration is 14 days. tool_name tool.name print(fPOLICY CHECK: Intercepting {tool_name}) if tool_name update-hotel and checkin_date in args and checkout_date in args: start datetime.fromisoformat(args[checkin_date]) end datetime.fromisoformat(args[checkout_date]) duration (end - start).days if duration 14: print(BLOCKED: Stay too long) return {result: Error: Maximum stay duration is 14 days.} return None # Post processing async def enrich_response( tool: ToolboxTool, args: Dict[str, Any], tool_context: ToolContext, tool_response: Any, ) - Optional[Any]: Callback fired after a tool execution. Enriches response for successful bookings. if isinstance(tool_response, dict): result tool_response.get(result, ) elif isinstance(tool_response, str): result tool_response else: return None tool_name tool.name if isinstance(result, str) and Error not in result: if tool_name book-hotel: loyalty_bonus 500 enriched_result fBooking Confirmed!\n You earned {loyalty_bonus} Loyalty Points with this stay.\n\nSystem Details: {result} if isinstance(tool_response, dict): modified_response deepcopy(tool_response) modified_response[result] enriched_result return modified_response else: return enriched_result return None接入点在创建Agent时把两个回调分别传给before_tool_callback和after_tool_callbacktoolset ToolboxToolset( server_urlhttp://127.0.0.1:5000, toolset_namemy-toolset, credentialsCredentialStrategy.toolbox_identity(), ) tools await toolset.get_tools() root_agent Agent( nameroot_agent, modelgemini-3-flash-preview, instructionSYSTEM_PROMPT, toolstools, # add any pre and post processing callbacks before_tool_callbackenfore_business_rules, after_tool_callbackenrich_response, )注意示例中的返回语义预处理回调返回一个带result的字典即错误信息时该轮工具调用被业务规则拦截返回None表示放行。后处理回调返回修改后的响应对象对不满足条件的情况返回None保持原响应不变。模型名gemini-3-flash-preview是文档示例使用的值按你的环境替换为实际可用的模型。示例还包含系统提示词、Runner会话初始化和两轮测试对话Book hotel with id 3. 触发后处理Update hotel with id 5 with checkin date 2025-01-18 and checkout date 2025-02-10 触发预处理拦截完整内容以仓库中的 agent.py 为准。Python LangChain用 wrap_tool_call 中间件实现同一目标如果你的智能体基于 LangChain文档给出的替代路径是ToolboxClient加 LangChain 中间件。依赖版本来自 requirements.txtpip install langchain1.3.9 langchain-google-genai4.2.1 toolbox-langchain1.0.0与 ADK 的关键差异在于LangChain 用wrap_tool_call装饰的中间件同时包裹工具调用预处理逻辑写在handler(request)之前后处理逻辑写在之后。完整示例见 agent.py核心结构如下# Pre processing wrap_tool_call async def enforce_business_rules(request, handler): Business Logic Validation: Enforces max stay duration (e.g., max 14 days). tool_call request.tool_call name tool_call[name] args tool_call[args] print(fPOLICY CHECK: Intercepting {name}) if name update-hotel: if checkin_date in args and checkout_date in args: try: start datetime.fromisoformat(args[checkin_date]) end datetime.fromisoformat(args[checkout_date]) duration (end - start).days if duration 14: print(BLOCKED: Stay too long) return ToolMessage( contentError: Maximum stay duration is 14 days., tool_call_idtool_call[id], ) except ValueError: pass # Ignore invalid date formats # PRE: Code here runs BEFORE the tool execution # EXEC: Execute the tool (or next middleware) result await handler(request) # POST: Code here runs AFTER the tool execution return result# Post processing wrap_tool_call async def enrich_response(request, handler): Post-Processing Enrichment: Adds loyalty points information to successful bookings. Standardizes output format. # PRE: Code here runs BEFORE the tool execution # EXEC: Execute the tool (or next middleware) result await handler(request) # POST: Code here runs AFTER the tool execution if isinstance(result, ToolMessage): content str(result.content) tool_name request.tool_call[name] if tool_name book-hotel and Error not in content: loyalty_bonus 500 result.content fBooking Confirmed!\n You earned {loyalty_bonus} Loyalty Points with this stay.\n\nSystem Details: {content} return resultasync with ToolboxClient(http://127.0.0.1:5000) as client: tools await client.aload_toolset(my-toolset) model ChatGoogleGenerativeAI(modelgemini-3-flash-preview) agent create_agent( system_promptsystem_prompt, modelmodel, toolstools, # add any pre and post processing methods middleware[enforce_business_rules, enrich_response], )与 ADK 的返回方式不同LangChain 中拦截时直接返回一个ToolMessage带tool_call_id工具执行会被短路错误信息直接作为工具结果进入对话。JavaScript 实现beforeToolCallback/afterToolCallback 与 createMiddleware文档同时提供 JS 版本两种框架各一个示例ADK.jsagent.js依赖google/adk^1.2.0与toolbox-sdk/adk^0.3.0在LlmAgent上直接挂beforeToolCallback和afterToolCallback通过toolbox-sdk/adk的ToolboxClient调用loadToolset(my-toolset)获取工具。注意示例中GOOGLE_GENAI_API_KEY从环境变量GOOGLE_API_KEY读取文档注释要求替换为你自己的 API key回调返回undefined表示放行返回错误字符串表示拦截。LangChain.jsagent.js依赖toolbox-sdk/core^1.0.0、langchain^1.2.25用createMiddleware的wrapToolCall钩子实现同样的执行前校验 执行后增强拦截时返回ToolMessage({content: ..., status: error})。注意该示例需要先通过langchain/core/tools的tool()把 Toolbox 工具包装成 LangChain 工具格式取getName()、getDescription()、getParamSchema()。npm start两个示例目录的package.json都定义了start脚本node agent.js依赖按各自目录的package.json安装即可。运行与验证结果各示例内置了两轮测试对话第一轮执行book-hotel验证后处理增强第二轮执行超 14 天的update-hotel验证预处理拦截。文档给出的期望输出如下AI: Booking Confirmed! You earned 500 Loyalty Points with this stay. AI: Error: Maximum stay duration is 14 days.文档明确提示由于 LLM 的非确定性以及不同编排框架之间的差异实际响应可能与之不完全一致以上应作为文档示例而非固定预期。除 AI 回复外示例代码在拦截发生时还会打印POLICY CHECK: Intercepting update-hotel和BLOCKED: Stay too long这类日志标记可以据此判断中间件确实被触发。文档给出的配套实践与限制文档在最佳实践部分还给出了几条与工具级中间件直接相关的做法可按需采用结构化日志用带关联 ID 的结构化 JSON 日志代替简单 print以便追踪一次用户请求穿越多个智能体轮次和工具调用。可测试性日志LLM 响应是非确定的可能概括掉关键细节。文档建议在中间件中加入显式日志标记例如logger.info(ACTION_SUCCESS: id)让集成测试通过 grep 这些稳定标记来验证工具是否成功而不是解析多变的自然语言响应。token 经济工具常返回冗长 JSON可在后处理中裁剪无用字段或先汇总大结果再送入 LLM 上下文。缓存对读多写少的工具如search_knowledge_base实现缓存中间件相同查询直接返回既有结果。错误处理在中间件捕获工具异常并返回结构化错误信息例如Error: Database timeout, please try again格式良好的错误信息能让 LLM 理解失败原因并自动用修正后的参数重试。安全侧的限制也要留意文档建议遵循最小权限原则把只读模式、敏感操作前的身份校验放在中间件中执行并在记录日志前主动剥离参数中的 PII。需要记住的边界是钩子和中间件机制本身由编排框架提供ADK 的回调、LangChain 的 middleware/wrap 钩子Toolbox SDK 只负责把工具集暴露给这些框架。如果你的框架不在文档覆盖范围内可以先按 Python 或 JS 示例的模式对照所用框架自身的钩子文档做等价实现。【免费下载链接】mcp-toolboxMCP Toolbox for Databases is an open source MCP server for databases.项目地址: https://gitcode.com/GitHub_Trending/ge/mcp-toolbox创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考