![[MCP在LangChain中的应用-04]利用ToolCallInterceptor将AOP引入工具调用](http://pic.xiahunao.cn/yaotu/[MCP在LangChain中的应用-04]利用ToolCallInterceptor将AOP引入工具调用)
MCP服务器作为独立进程运行无法访问LangGraph运行时信息比如状态、长期存储和静态上下文等。ToolCallInterceptor弥补了这一缺陷它使我们能够在MCP工具执行期间可用读取这些运行时信息。ToolCallInterceptor提供类似中间件的工具调用控制可以修改请求、实现重试、动态添加HTTP报头或“短路终止”执行。1. ToolCallInterceptor协议如下面的代码片段所示ToolCallInterceptor作为一个协议只定义了一个__call__方法意味着我们可以将拦截器定义为函数。这个__call__方法也采用中间件的定义风格它利用request参数表示的MCPToolCallRequest对象作为请求通过调用handler参数代表的可执行对象将请求交付给后续的处理流程并得到工具调用结果的MCPToolCallResult对象。由于拦截器可以自主决定是否调用handler,或者在何时调用handler所以它不仅可以在工具调用前修改请求在调用后修改结果还可以直接生成工具调用的结果。我们利用__init__方法的tool_interceptors参数将ToolCallInterceptor列表注册到MultiServerMCPClient对象上。classToolCallInterceptor(Protocol):asyncdef__call__(self,request:MCPToolCallRequest,handler:Callable[[MCPToolCallRequest],Awaitable[MCPToolCallResult]],)-MCPToolCallResultclassMultiServerMCPClient:def__init__(self,connections:dict[str,Connection]|NoneNone,*,callbacks:Callbacks|NoneNone,tool_interceptors:list[ToolCallInterceptor]|NoneNone,tool_name_prefix:boolFalse,)-None2. 作为请求的MCPToolCallRequest如下所示的表示工具调用请求的MCPToolCallRequest类型。dataclassclassMCPToolCallRequest:name:strargs:dict[str,Any]server_name:strheaders:dict[str,Any]|NoneNoneruntime:object|NoneNonedefoverride(self,**overrides:Unpack[_MCPToolCallRequestOverrides])-MCPToolCallRequestclass_MCPToolCallRequestOverrides(TypedDict,totalFalse):name:NotRequired[str]args:NotRequired[dict[str,Any]]headers:NotRequired[dict[str,Any]|None]类型成员说明如下name工具名称args调用工具的参数字典server_nameMCP服务器名称headersHTTP请求报头runtime: LangGraph运行时override用于修改请求携带的工具名称、参数字典和HTTP请求报头3. 作为结果的MCPToolCallResult表示工具调用结果的类型MCPToolCallResult定义如下它可以是一个mcp.types.CallToolResult对象、ToolMessage消息或者一个Command命令。ToolMessage和Command是LangChain对工具执行结果的表达所以只有在ToolCallInterceptor链最终的返回类型为CallToolResult的情况下采用将其转换成ToolMessage类型。MCPToolCallResultCallToolResult|ToolMessage|CommandclassCallToolResult(Result):content:list[ContentBlock]structuredContent:dict[str,Any]|NoneNoneisError:boolFalse4. 实例演示利用ToolCallInterceptor从HTTP请求报头中提取用户ID前面为了说明执行工具返回的ToolMessage是如何根据返回的ToolCallResult生成的我们构建了一个MCP服务器现在我们对它略加改动来演示ToolCallInterceptor的应用。如下面的代码片段所示我们将工具函数get_profile的user_id参数去掉替换它的request是一个自动注入的参数表示当前HTTP请求我们从请求的x-user-id报头提取用户ID。fromfastmcpimportFastMCPfromstarlette.requestsimportRequestfromfastmcp.dependenciesimportCurrentRequest mcpFastMCP(Server)mcp.tool()asyncdefget_profile(request:RequestCurrentRequest())-str:Get user profile information.user_idrequest.headers.get(x-user-id)print(fReceived request for user_id:{user_id})ifuser_id:return(fMy name is John Doe (user id is{user_id}), and I am a software engineer with 5 years of experience in web development.I enjoy working with Python and JavaScript, and I have a passion for learning new technologies.In my free time, I like to travel and explore new places.)else:raiseValueError(User ID is required to get profile information.)mcp.run(transportstreamable-http,host0.0.0.0)在如下所示的客户端程序中我们调用create_agent函数创建了一个使用ChatOpenAI作为模型的Agent注册的工具通过连接上面这个MCP服务器的MultiServerMCPClient进行提取。我们通过context_schema参数设置了静态上下文Schema类型Context其中包含我们所需的用户ID。fromtypingimportcast,TypedDictfromlanggraph.runtimeimportRuntimefromlangchain_mcp_adapters.clientimportMultiServerMCPClientfromlangchain_mcp_adapters.interceptorsimportMCPToolCallRequestfromlangchain.agentsimportcreate_agentfromdotenvimportload_dotenvfromlangchain_openaiimportChatOpenAIimportasyncio load_dotenv()classContext(TypedDict):user_id:strasyncdefprofile_interceptor(request:MCPToolCallRequest,handler):runtime:Runtime[Context]cast(Runtime[Context],request.runtime)requestrequest.override(headers{**(request.headersor{}),x-user-id:runtime.context[user_id]})resultawaithandler(request)returnresultasyncdefmain():clientMultiServerMCPClient(connections{greeting:{transport:streamable_http,url:http://localhost:8000/mcp}},tool_interceptors[profile_interceptor])agentcreate_agent(modelChatOpenAI(modelgpt-5.2-chat),toolsawaitclient.get_tools(),context_schemaContext,)resultawaitagent.ainvoke(input{messages:[{type:user,content:What is the profile information for current user?}]},context{user_id:123},)print(result[messages][-1].content)asyncio.run(main())MultiServerMCPClient注册了一个通过profile_interceptor函数表示的ToolCallInterceptor我们利用它在请求中添加表示用户ID的x-user-id报头。具体做法很简单我们利用request参数表示的MCPToolCallRequest对象得到表示LangGragh运行时的Runtime[Context]对象并从作为静态上下文的Context对象中提取出用户ID。最后我们通过调用request的override方法在返回的新请求中添加了所需的x-user-id报头。最后我们指定提示词What is the profile information for current user?调用Agent最终会得到如下的结果Here is the profile information for current user: - **Name:** John Doe - **Profession:** Software Engineer - **Experience:** 5 years in web development - **Skills Interests:** Enjoys working with Python and JavaScript; passionate about learning new technologies - **Hobbies:** Traveling and exploring new places