LangChain--10--MCP

提示文章写完后目录可以自动生成如何生成可参考右边的帮助文档文章目录MCP--模型上下文协议1. MCP介绍2. LangChain Agent集成MCP2.1 传输方式2.2. 创建MCP服务器2.3. Agent配置MCP工具2.4. LangChain Agent Tools与MCP Tools 区别3. MCP认证与安全4. 工具拦截器Tool Interceptor5. MCP 核心特性MCP–模型上下文协议RESTful–介绍1. MCP介绍2. LangChain Agent集成MCPpipinstallfastmcp3.4.2 pipinstalllangchain-mcp-adapters0.3.02.1 传输方式关于Stdio更多可配参数参考https://reference.langchain.com/python/langchain-mcp-adapters/sessions/StdioConnection...... clientMultiServerMCPClient({my_server:{transport:http,url:http://localhost:8000/mcp,headers:{Authorization:Bearer YOUR_TOKEN,},}})toolsawait client.get_tools()......关于HTTP模式更多可配参数参考https://reference.langchain.com/python/langchain-mcp-adapters/sessions/StreamableHttpConnection2.2. 创建MCP服务器如下代码是创建stdio传输版本——本地工具服务math_server.py 客户端以子进程方式启动此脚本通过标准输入/输出通信 fromfastmcpimportFastMCP# 创建一个 MCP Server 实例名为 MathmcpFastMCP(Math)mcp.tool()defadd(a:int,b:int)-int:将两个数字相加returnabmcp.tool()defmultiply(a:int,b:int)-int:将两个数字相乘returna*bif__name____main__:# 以 stdio 模式启动mcp.run(transportstdio)如下代码是创建HTTP传输版本——远程天气服务(weather_server.py) 在 localhost:8000/mcp 启动一个 Web 服务 fromfastmcpimportFastMCP# 创建一个 MCP 服务, 服务名称为 WeathermcpFastMCP(Weather)mcp.tool()asyncdefget_weather(location:str)-str:获取指定城市的天气信息# 实际项目中应调用外部天气 APIweather_data{北京:晴22~28°C北风3级,上海:多云25~30°C东南风2级,深圳:阵雨26~31°C南风4级,}returnweather_data.get(location,f暂未找到{location}的天气数据)if__name____main__:# 以 streamable-http 模式启动监听 8000 端口mcp.run(transportstreamable-http)2.3. Agent配置MCP工具 LangChain Agent 集成 MCP 同时连接数学工具和天气工具两个 MCP Server importasynciofromlangchain_mcp_adapters.clientimportMultiServerMCPClientfromlangchain.agentsimportcreate_agentfrominit_llmimportdeepseek_llmasyncdefmain():# 1. 创建 MCP 客户端——通过字典配置多个 Server 的连接信息clientMultiServerMCPClient({math:{transport:stdio,# stdio 本地子进程通信command:D:/ProgramData/miniconda3/envs/langchain_v1.2/python.exe,# 指定所使用的 Python 解释器路径args:[D:/PyCharmSpace/LangChainV1.2Project/08_mcp/quick_start/math_server.py],# MCP Server 脚本的绝对路径env:{# 传给子进程的环境变量确保 UTF-8 编码PYTHONIOENCODING:utf-8,},},weather:{transport:http,# HTTP 远程通信url:http://localhost:8000/mcp,},})# 2. 从所有已配置的 Server 中加载工具转换成 LangChain BaseTool 列表toolsawaitclient.get_tools()print(f加载到的工具:{tools})# 3. 像使用普通 LangChain Tool 一样传入 Agentagentcreate_agent(modeldeepseek_llm,toolstools,)# 4. Agent 调用时MCP 工具和普通 LangChain 工具无区别math_responseawaitagent.ainvoke({messages:[{role:user,content:计算 (3 5) × 12 的结果}]})weather_responseawaitagent.ainvoke({messages:[{role:user,content:上海今天的天气怎么样}]})print(math_response,math_response)print(weather_response,weather_response)print(f数学结果:{math_response[messages][-1].content})print(f天气结果:{weather_response[messages][-1].content})if__name____main__:asyncio.run(main())2.4. LangChain Agent Tools与MCP Tools 区别3. MCP认证与安全fromfastmcp.server.auth.providers.jwtimportRSAKeyPair# 1.生成 RSA 密钥对key_pairRSAKeyPair.generate()private_keykey_pair.private_key.get_secret_value()# 私钥用于签发 Token保密public_keykey_pair.public_key# 公钥用于验证 Token可公开# 2.用私钥签发 JWT Token create_token方法内已包含私钥的使用jwt_tokenkey_pair.create_token(subjectagent-011,# sub: Token 代表的主体这里是 Agent 的 IDissuermy-company-auth-server,# iss: 签发方标识Server 会校验此值audienceinternal-mcp-server,# aud: 接收方标识表明 Token 是专门给这个 MCP Server 用的expires_in_seconds3600,# exp: Token 1 小时后过期)# 3.输出结果print(*70)print(公钥后续复制到 mcp_server.py 的 PUBLIC_KEY 变量中)print(*70)print(public_key)print()print(*70)print(JWT Token后续复制到 mcp_client.py 的 JWT_TOKEN 变量中)print(*70)print(jwt_token)print() MCP Server 服务端 所有请求必须携带有效的 JWT Token否则返回 401。 fromfastmcpimportFastMCPfromfastmcp.server.authimportJWTVerifier# 1. 配置将 gen_token.py 输出的公钥内容完整粘贴到这里包含 BEGIN/END 行PUBLIC_KEY-----BEGIN PUBLIC KEY----- MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAs/WavxnBoIX/sclvsdv3 IFrz09nDCWgnWQUZ581/tgYt4uE7zrFM0qzhwEZvuApRtG/jdI7fm6pkFMUzDy QilXk1zoiMe8LKSSIFWVYZ82HNsIreii4WSjeLfjtbNSSoUUOPWNUFtFUCjyd3c DvSadOIH3UcQYvaFvLn0Np5kljPmVMcn82FaCOudXDoKtn/z0Gqca1Yc6RCb8Zzz seE0or8AjYsLeBWWhJEa2vRvXAvGVt2AEzwbvvbVQYnKs/kOhZIt/01ROfzeqS zKrQXSVjizwp5dPY7IwnBeEEMFUrfEPsBKaCOO4ed8z79PlNihdADjIbpOTUjMtA KwIDAQAB -----END PUBLIC KEY-----# 2. 创建 JWTVerifier 认证提供方authJWTVerifier(public_keyPUBLIC_KEY,# 用这个公钥验证 JWT 签名issuermy-company-auth-server,# 只接受由此签发方签发的 Token这里要与 gen_token.py 中的 issuer 一致audienceinternal-mcp-server,# 只接受签发给本服务的 Token这里要与 gen_token.py 中的 audience 一致algorithmRS256,# 签名算法需与签发时一致)# 3. 创建带认证的 MCP 服务器mcpFastMCP(Internal Tools Server,authauth)# 模拟的企业数据EMPLOYEE_DB{E001:{name:张三,department:技术部,position:高级工程师},E002:{name:李四,department:财务部,position:财务经理},E003:{name:王五,department:市场部,position:市场总监},}BUDGET_DB{技术部:{total:5000000,used:3200000,remaining:1800000},财务部:{total:1500000,used:900000,remaining:600000},市场部:{total:3000000,used:2100000,remaining:900000},}# 4. 定义受保护的工具mcp.tool()asyncdefquery_employee(employee_id:str)-str:查询员工基本信息。 参数: employee_id: 员工编号例如 E001 employeeEMPLOYEE_DB.get(employee_id)ifemployeeisNone:returnf未找到员工编号为{employee_id}的记录return(f员工信息姓名{employee[name]}f部门{employee[department]}f职位{employee[position]})mcp.tool()asyncdefquery_department_budget(department:str)-str:查询部门预算信息。 参数: department: 部门名称例如 技术部 budgetBUDGET_DB.get(department)ifbudgetisNone:returnf未找到部门 {department} 的预算记录return(f部门预算总额{budget[total]/10000}万f已用{budget[used]/10000}万f剩余{budget[remaining]/10000}万)# 5. 启动服务器if__name____main__:mcp.run(transporthttp,host127.0.0.1,port8000) 客户端只需将 Token 放入 headers 即可完成认证无需理解 JWT 内部结构。 importasynciofromlangchain_mcp_adapters.clientimportMultiServerMCPClientfromlangchain.agentsimportcreate_agentfrominit_llmimportdeepseek_llm# 1. 将从 gen_token.py 输出的 JWT Token 粘贴到这里JWT_TOKENeyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiJ9.eyJzdWIiOiJhZ2VudC0wMTEiLCJpc3MiOiJteS1jb21wYW55LWF1dGgtc2VydmVyIiwiaWF0IjoxNzgzOTUzNDYzLCJleHAiOjE3ODM5NTcwNjMsImF1ZCI6ImludGVybmFsLW1jcC1zZXJ2ZXIifQ.iGqolffc1OGVCqDQKGAVhdHW5VPCQW-TCRbAY6_haN0K3peuRRB0jsOXSa9wpj-2jzCIii9hq6xYTDvVMd4Fw4sAKQFKeTzGr-g0cYw0kJZcTaVxhHzwE48pQAYJAZHkGJtUo2cT0-qm87ol4uZTpz5JtXahZsOjbSQaZLctKhn3jmdguY_l1WfxRhdz5AUibx461tZ4s6l7xqMy4LmAsRxG15f_jtjXF5CL6UxJMAARQujQrG-jloPlv3HyyV5EgWagT1TxzLbqZPVNfCnxu1uK8SIFLu2H0IFko3osd597QZkUQ62ONuC-WmCVcobzuZZu_dIBvYxOPYI6Q9jWwA# 2. 创建 MCP 客户端通过 headers 携带 Bearer Tokenasyncdefmain():clientMultiServerMCPClient({internal_tools:{transport:http,url:http://localhost:8000/mcp,# 将 Token 放入 Authorization 请求头headers:{Authorization:fBearer{JWT_TOKEN},},}})# 获取工具列表并创建 Agenttoolsawaitclient.get_tools()agentcreate_agent(deepseek_llm,tools)# 场景 1查询员工信息responseawaitagent.ainvoke({messages:[{role:user,content:帮我查一下员工 E001 的信息}]})print(fAgent 回复:{response[messages][-1].content})print(*50)# 场景 2查询部门预算responseawaitagent.ainvoke({messages:[{role:user,content:技术部的预算使用情况如何}]})print(fAgent 回复:{response[messages][-1].content})if__name____main__:asyncio.run(main())4. 工具拦截器Tool Interceptormatch_server.py:fromfastmcpimportFastMCP mcpFastMCP(MathService)mcp.tool()defadd(a:int,b:int)-int:计算两个整数的和returnabmcp.tool()defmultiply(a:int,b:int)-int:计算两个整数的积returna*bif__name____main__:mcp.run(transporthttp,host127.0.0.1,port8000)interceptor_demo1.py:importasynciofromlangchain_mcp_adapters.clientimportMultiServerMCPClientfromlangchain_mcp_adapters.interceptorsimportMCPToolCallRequestfromlangchain.agentsimportcreate_agentfrominit_llmimportdeepseek_llm# 定义一个日志拦截器 asyncdeflogging_interceptor(request:MCPToolCallRequest,handler):在每次 MCP 工具调用前后打印日志print(request:,request)print(f[拦截器] 调用前 —— 工具:{request.name}, 参数:{request.args})resultawaithandler(request)# 调用下一个拦截器或实际工具print(f[拦截器] 调用后 —— 工具:{request.name}, 返回:{result})returnresultasyncdefmain():# 创建 MCP 客户端通过 tool_interceptors 参数传入拦截器clientMultiServerMCPClient({math:{transport:http,url:http://localhost:8000/mcp,},},tool_interceptors[logging_interceptor],# 传入拦截器列表)toolsawaitclient.get_tools()print(f已加载工具:{[t.namefortintools]})agentcreate_agent(modeldeepseek_llm,toolstools)responseawaitagent.ainvoke({messages:[{role:user,content:请计算 8 乘以 5 的结果}]})print(fAgent 最终回复:{response[messages][-1].content})if__name____main__:asyncio.run(main())asyncdefouter(request,handler):print(外层: before)resultawaithandler(request)print(外层: after)returnresultasyncdefinner(request,handler):print(内层: before)resultawaithandler(request)print(内层: after)returnresult# tool_interceptors[outer, inner] 的执行顺序# 外层: before → 内层: before → 工具执行 → 内层: after → 外层: after 多个工具拦截器执行顺序 importasynciofromlangchain_mcp_adapters.clientimportMultiServerMCPClientfromlangchain.agentsimportcreate_agentfrominit_llmimportdeepseek_llm# 定义一个日志拦截器 asyncdefouter(request,handler):print(外层: before)resultawaithandler(request)print(外层: after)returnresultasyncdefinner(request,handler):print(内层: before)resultawaithandler(request)print(内层: after)returnresultasyncdefmain():# 创建 MCP 客户端通过 tool_interceptors 参数传入拦截器clientMultiServerMCPClient({math:{transport:http,url:http://localhost:8000/mcp,},},tool_interceptors[outer,inner],# 传入拦截器列表 , 执行顺序为 outer → inner)toolsawaitclient.get_tools()print(f已加载工具:{[t.namefortintools]}\n)agentcreate_agent(modeldeepseek_llm,toolstools)responseawaitagent.ainvoke({messages:[{role:user,content:请计算 3 加 5 的结果}]})print(fAgent 最终回复:{response[messages][-1].content})if__name____main__:asyncio.run(main())5. MCP 核心特性