系列前文 GitHub Actions 深度实践零运维搭建 CI/CD 流水线本文定位在「流水线能自动构建/发布」之后解决下一层问题——AI Agent 如何标准化连接外部工具。技术栈Python 3.10 · 官方mcpSDKFastMCP·langchain-mcp-adapters·可选GitHub Actions 做门禁MCPModel Context Protocol正在成为 AI Agent 连接外部工具的事实标准。它解决了传统 AI 集成的m×n 困境m 个大模型 × n 个工具往往要做 m×n 次适配MCP 将其降维为mn——模型侧实现一次 Client工具侧实现一次 Server即可全互联。本文将通过可运行的实战代码从零构建一套完整的 MCP 工具链并说明每个关键技术的应用价值。一、理解 MCP 的核心架构MCP 采用 Host / Client / Server 三层┌─────────────────────────────────────────────┐ │ Host宿主 │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ Client A │ │ Client B │ │ Client C │ │ │ └────┬─────┘ └────┬─────┘ └────┬─────┘ │ └────────┼─────────────┼─────────────┼────────┘ │ │ │ ┌─────▼─────┐ ┌─────▼─────┐ ┌─────▼─────┐ │ MCP Server│ │ MCP Server│ │ MCP Server│ │ 数据库 │ │ 文件系统│ │ 天气 API│ └───────────┘ └───────────┘ └───────────┘角色职责类比Host运行 AI 的应用Claude Desktop、自研 Agent操作系统ClientHost 内组件维持与某个 Server 的独立会话驱动Server暴露 Tools / Resources / Prompts外设与服务底层通信基于JSON-RPC 2.0。本地开发常用stdio进程管道远程/多客户端场景常用Streamable HTTP / SSE便于 Server → Client 推送。应用价值工具开发与 Agent 开发彻底解耦。业务方只维护一个 MCP ServerCursor、Claude Desktop、LangChain Agent 都能按同一协议接入不必为每个宿主重写插件。二、环境准备2.1 Python 环境# 创建虚拟环境 python -m venv mcp-agent-env # Windows mcp-agent-env\Scripts\activate # macOS / Linux # source mcp-agent-env/bin/activate pip install mcp[cli] httpx pydantic langchain-openai langchain-mcp-adapters langgraphWindows 下命令一般是python不是python3。下文示例统一写python按本机解释器调整即可。2.2 项目结构mcp-toolchain/ ├── mcp_server.py # MCP Server暴露工具 ├── agent_host.py # 最小 Host发现/调用工具 ├── langchain_integration.py # LangChain Agent 集成 ├── .github/workflows/ci.yml # 可选用 Actions 做门禁 └── requirements.txt与系列前文的对应关系前文Actions本文MCPWorkflow 发布过程代码化Server 工具能力代码化Secrets 管密钥Token / 权限在 Server 层校验失败可观测 ArtifactTool 调用审计日志 限流三、构建 MCP Server暴露工具能力用官方FastMCP实现天气查询、知识库检索、网页摘要三个工具。# mcp_server.py import json import httpx from mcp.server.fastmcp import FastMCP mcp FastMCP(ResearchAssistantTools) # Tool 1: 天气查询 mcp.tool() async def get_weather(city: str) - str: 查询指定城市的实时天气信息。 Args: city: 城市名称例如 Beijing、Shanghai async with httpx.AsyncClient() as client: response await client.get( fhttps://wttr.in/{city}, params{format: j1}, timeout10.0, ) response.raise_for_status() data response.json() current data[current_condition][0] return json.dumps( { city: city, temperature_c: current[temp_C], feels_like_c: current[FeelsLikeC], humidity: current[humidity], description: current[weatherDesc][0][value], }, ensure_asciiFalse, ) # Tool 2: 知识库检索 KNOWLEDGE_BASE { MCP协议: ( MCPModel Context Protocol是开放标准协议 用于让 AI 应用以统一方式连接外部工具与数据源。 ), AI Agent: ( AI Agent智能体能感知环境、决策并调用工具完成任务 工具调用通常通过 Function Calling 或 MCP 完成。 ), } mcp.tool() async def search_knowledge(query: str) - str: 从本地知识库中检索相关信息。 results [] q query.lower() for key, value in KNOWLEDGE_BASE.items(): if q in key.lower() or q in value.lower(): results.append({topic: key, content: value}) if not results: return json.dumps( {status: no_result, message: f未找到与{query}相关的知识}, ensure_asciiFalse, ) return json.dumps({status: success, results: results}, ensure_asciiFalse) # Tool 3: 网页摘要预览 mcp.tool() async def summarize_url(url: str) - str: 获取指定 URL 的网页内容预览生产环境应再接摘要模型。 async with httpx.AsyncClient() as client: response await client.get(url, timeout15.0, follow_redirectsTrue) content response.text[:3000] return json.dumps( { url: url, status_code: response.status_code, content_preview: content[:1500], char_count: len(response.text), }, ensure_asciiFalse, ) if __name__ __main__: # stdio适合本地 Host 拉起子进程远程场景可改为 streamable-http / sse mcp.run(transportstdio)关键技术与价值技术点作用应用价值mcp.tool()把函数注册为可发现工具docstring / 类型注解自动变成 Agent 可理解的 schemaasynchttpx非阻塞外呼多工具并发时不堵死事件循环transportstdio标准输入输出通信零端口、零 CORS本地调试最快返回 JSON 字符串统一结果载体Host / 多模型侧解析成本低对比「每个平台写一套 Plugin」同等三个工具传统适配往往要几百行胶水FastMCP 声明式注册后维护点收敛到一个 Server 进程。本地自检可选# 若已安装 mcp CLI mcp dev mcp_server.py四、构建 MCP Host连接 Server 并驱动调用Host 是「能力消费者」通过 MCP Client 发现工具、发起tools/call。# agent_host.py import asyncio import logging from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client logging.basicConfig( levellogging.INFO, format%(asctime)s - HOST - %(levelname)s - %(message)s, ) async def main(): server_params StdioServerParameters( commandpython, args[mcp_server.py], ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() logging.info(Host 已连接到 MCP Server) # 1. 发现工具 listed await session.list_tools() print(可用工具:, [t.name for t in listed.tools]) # 2. 调用天气 weather await session.call_tool(get_weather, {city: Beijing}) print(天气:, weather.content) # 3. 调用知识库 knowledge await session.call_tool( search_knowledge, {query: MCP协议} ) print(知识:, knowledge.content) if __name__ __main__: asyncio.run(main())运行python agent_host.py关键技术与价值技术点作用应用价值StdioServerParameters声明如何拉起 Server 子进程Host 不关心工具内部实现只关心命令与参数list_tools运行时能力发现新增mcp.tool后 Host 无需改协议代码call_toolJSON-RPCtools/call封装调用面统一便于做审计、重试、熔断双层async with正确关闭管道与会话避免僵尸子进程占满文件句柄这一层还不接大模型——先保证「发现 → 调用 → 拿结果」稳定再接到 Agent排障成本低一个数量级。五、LangChain 集成可治理的 Agent原则LangChain / LangGraph 是「大脑」MCP 是「手脚」。所有外呼走 MCP Tool权限与限流放在 Server 侧避免模型「直接摸」任意 HTTP。使用官方适配库langchain-mcp-adapters不要再用虚构的langchain.tools.mcpAPI。# langchain_integration.py import asyncio import logging import time from langchain.agents import create_agent from langchain_mcp_adapters.client import MultiServerMCPClient from langchain_mcp_adapters.tools import load_mcp_tools logging.basicConfig(levellogging.INFO) logger logging.getLogger(audit) async def main(): client MultiServerMCPClient( { research: { command: python, args: [mcp_server.py], transport: stdio, } } ) # 有状态会话同一 Server 连接上复用适合多步工具编排 async with client.session(research) as session: tools await load_mcp_tools(session) logger.info(已加载工具: %s, [t.name for t in tools]) agent create_agent( modelopenai:gpt-4o, toolstools, system_prompt( 你是智能研究助手只能通过已提供的工具查询天气、 检索知识库、获取网页预览。不要编造工具结果。 ), ) result await agent.ainvoke( { messages: [ { role: user, content: 北京今天天气怎么样另外简要解释 MCP 协议是什么。, } ] } ) print(最终回答:, result[messages][-1].content) if __name__ __main__: asyncio.run(main())需要 OpenAI 兼容密钥时# Windows PowerShell $env:OPENAI_API_KEYsk-... python langchain_integration.py5.1 审计回调可观测若你使用带 callback 的 Executor / 自定义图节点建议输出结构化步骤日志class AuditCallbackHandler: 示意记录工具名与入参便于审计与排障。 def on_tool_start(self, tool_name: str, tool_input: dict): logging.info( agent_step, extra{ tool: tool_name, input: tool_input, timestamp: time.time(), }, )关键技术与价值技术点作用应用价值MultiServerMCPClient同时挂多个 MCP Server天气、知识库、内部 API 可分进程拆分互不影响load_mcp_toolsMCP Tool → LangChain ToolAgent 框架与工具协议解耦换大脑不必重写手脚create_agent工具调用 Agent多步任务由模型编排人只写工具与策略审计日志记录每步 tool / input满足内控「谁在何时调了什么」线上事故可回放六、进阶JSON-RPC 与 SSE / Streamable HTTP6.1 JSON-RPC 2.0 请求{ jsonrpc: 2.0, id: 1, method: tools/call, params: { name: get_weather, arguments: { city: Beijing } } }响应{ jsonrpc: 2.0, id: 1, result: { content: [{ type: text, text: ... }] } }应用价值抓包/日志按id对齐请求与响应自研 Host 时不必发明私有 RPC。6.2 SSE / Streamable HTTP远程场景远程部署时Server 可改为 HTTP 传输具体路径以当前 SDK 版本文档为准典型事件流形态Content-Type: text/event-stream Cache-Control: no-cache Connection: keep-alive event: message data: {jsonrpc:2.0,id:1,result:{tools:[...]}}应用价值多 Host 共享同一工具集群与 K8s Service / 网关鉴权更好接。本地仍推荐stdio降低复杂度。把 Server 改成 HTTP 传输后Client 配置示例client MultiServerMCPClient( { research: { url: http://127.0.0.1:8000/mcp, transport: http, # 或文档推荐的 streamable-http / sse } } )七、与 GitHub Actions 衔接给 MCP 工具链加门禁前文解决「代码怎么自动验证与发布」MCP Server 同样需要 CI否则 Agent 一上线就踩坏工具。# .github/workflows/mcp-ci.yml name: MCP Toolchain CI on: push: branches: [main] pull_request: jobs: smoke: runs-on: ubuntu-latest timeout-minutes: 10 steps: - uses: actions/checkoutv4 - uses: actions/setup-pythonv5 with: python-version: 3.12 cache: pip - name: Install run: | python -m pip install -U pip pip install mcp[cli] httpx pytest pytest-asyncio - name: Import list tools smoke run: | python - PY import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def smoke(): params StdioServerParameters(commandpython, args[mcp_server.py]) async with stdio_client(params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() names {t.name for t in tools.tools} assert {get_weather, search_knowledge, summarize_url} names print(smoke ok:, sorted(names)) asyncio.run(smoke()) PY应用价值PR 合入前证明「工具可发现、进程可拉起」避免只测 Python import、不测 MCP 握手的假绿。八、生产级最佳实践8.1 权限与安全在 Server 层def validate_token(token: str | None) - str: if not token or token ! demo-token: raise PermissionError(无效令牌) return user-demo def has_permission(user_id: str, action: str) - bool: return action in {weather, knowledge} mcp.tool() async def get_weather(city: str, token: str | None None) - str: 带权限校验的天气查询示意。 user_id validate_token(token) if not has_permission(user_id, weather): raise PermissionError(无权访问天气服务) # ... 原业务逻辑 return json.dumps({city: city, ok: True}, ensure_asciiFalse)更稳妥的做法是把身份放在HTTP Header / 网关而不是让模型自由填写token参数——上面仅作「Server 必须有权控」的示意。8.2 异常兜底手段做法价值超时httpxtimeout10外部 API 挂了不拖死 Agent步数上限Agent / Graph 限制迭代次数防止工具互调死循环烧 Token敏感操作返回pending_approval高风险动作人工确认后再执行幂等写操作带request_id重试不产生重复工单/重复扣费8.3 可观测性每次tools/call打结构化日志tool/latency_ms/ok/error_type与前文 Actions 的失败通知打通Server 健康检查失败 → Webhook 告警对外部依赖做熔断连续失败则短时拒绝避免雪崩九、总结对比对比维度传统方式MCP 方式适配成本m×nmn代码维护点各平台各一套 Plugin一个 Server多 Host 复用扩展性新工具要重写适配新mcp.tool即插即用可治理性依赖各宿主特性统一在 Server 做权限 / 审计 / 限流与 CI/CD难统一回归stdio 冒烟可直接挂进 Actions落地顺序建议1. FastMCP 暴露 13 个只读工具stdio 2. agent_host 验证 list_tools / call_tool 3. langchain-mcp-adapters 接到 Agent 4. 加权限、超时、审计 5. CI 冒烟 可选HTTP 远程部署MCP 的核心价值是零适配工具开发与 Agent 开发解耦「一次编写到处调用」。叠加上文的零运维流水线整条链路变成代码推送 → Actions 自动验证/发布 → MCP Server 对外暴露工具 → Agent 按协议发现并调用这才是「深度实践」想拼完整的一环而不是又一个只能 Demo 的脚本。系列续篇工具接上之后必须打开多步推理黑盒——见 可观测性深度实践零盲盒追踪 Agent 多步推理。参考Model Context Protocol 规范MCP Python SDK / FastMCPLangChain MCP 文档langchain-mcp-adapters系列前文GitHub Actions 深度实践零运维搭建 CI/CD 流水线