系列前文： GitHub Actions 深度实践：零运维搭建 CI/CD 流水线
本文定位：在「流水线能自动构建/发布」之后，解决下一层问题—— AI Agent 如何标准化连接外部工具。
技术栈：Python 3.10+ · 官方 mcp SDK（FastMCP）· langchain-mcp-adapters ·（可选）GitHub Actions 做门禁

        MCP（Model Context Protocol）正在成为 AI Agent 连接外部工具的事实标准。它解决了传统 AI 集成的 m×n 困境：m 个大模型 × n 个工具，往往要做 m×n 次适配；MCP 将其降维为 m+n——模型侧实现一次 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）	操作系统
Client	Host 内组件，维持与某个 Server 的独立会话	驱动
Server	暴露 Tools / Resources / Prompts	外设与服务

        底层通信基于 JSON-RPC 2.0。本地开发常用 stdio（进程管道）；远程/多客户端场景常用 Streamable HTTP / SSE，便于 Server → Client 推送。

        应用价值： 工具开发与 Agent 开发彻底解耦。业务方只维护一个 MCP Server；Cursor、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 langgraph

Windows 下命令一般是 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）	本文（MCP）
Workflow = 发布过程代码化	Server = 工具能力代码化
Secrets 管密钥	Token / 权限在 Server 层校验
失败可观测 + Artifact	Tool 调用审计日志 + 限流

---

三、构建 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(
            f"https://wttr.in/{city}",
            params={"format": "j1"},
            timeout=10.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_ascii=False,
        )


# ==================== Tool 2: 知识库检索 ====================
KNOWLEDGE_BASE = {
    "MCP协议": (
        "MCP（Model 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_ascii=False,
        )
    return json.dumps({"status": "success", "results": results}, ensure_ascii=False)


# ==================== Tool 3: 网页摘要（预览） ====================
@mcp.tool()
async def summarize_url(url: str) -> str:
    """获取指定 URL 的网页内容预览（生产环境应再接摘要模型）。"""
    async with httpx.AsyncClient() as client:
        response = await client.get(url, timeout=15.0, follow_redirects=True)
        content = response.text[:3000]
        return json.dumps(
            {
                "url": url,
                "status_code": response.status_code,
                "content_preview": content[:1500],
                "char_count": len(response.text),
            },
            ensure_ascii=False,
        )


if __name__ == "__main__":
    # stdio：适合本地 Host 拉起子进程；远程场景可改为 streamable-http / sse
    mcp.run(transport="stdio")

关键技术与价值

技术点	作用	应用价值
@mcp.tool()	把函数注册为可发现工具	docstring / 类型注解自动变成 Agent 可理解的 schema
async + httpx	非阻塞外呼	多工具并发时不堵死事件循环
transport="stdio"	标准输入输出通信	零端口、零 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(
    level=logging.INFO,
    format="%(asctime)s - HOST - %(levelname)s - %(message)s",
)


async def main():
    server_params = StdioServerParameters(
        command="python",
        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_tool	JSON-RPC tools/call 封装	调用面统一，便于做审计、重试、熔断
双层 async with	正确关闭管道与会话	避免僵尸子进程占满文件句柄

        这一层还不接大模型——先保证「发现 → 调用 → 拿结果」稳定，再接到 Agent，排障成本低一个数量级。

---

五、LangChain 集成：可治理的 Agent

原则：LangChain / LangGraph 是「大脑」，MCP 是「手脚」。

        所有外呼走 MCP Tool，权限与限流放在 Server 侧，避免模型「直接摸」任意 HTTP。

        使用官方适配库 langchain-mcp-adapters（不要再用虚构的 langchain.tools.mcp API）。

# 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(level=logging.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(
            model="openai:gpt-4o",
            tools=tools,
            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_KEY="sk-..."
python langchain_integration.py

5.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_tools	MCP Tool → LangChain Tool	Agent 框架与工具协议解耦，换大脑不必重写手脚
create_agent	工具调用 Agent	多步任务由模型编排，人只写工具与策略
审计日志	记录每步 tool / input	满足内控「谁在何时调了什么」；线上事故可回放

---

六、进阶：JSON-RPC 与 SSE / Streamable HTTP

6.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/checkout@v4

      - uses: actions/setup-python@v5
        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(command="python", 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_ascii=False)

        更稳妥的做法是把身份放在 HTTP Header / 网关，而不是让模型自由填写 token 参数——上面仅作「Server 必须有权控」的示意。

8.2 异常兜底

手段	做法	价值
超时	httpx timeout=10	外部 API 挂了不拖死 Agent
步数上限	Agent / Graph 限制迭代次数	防止工具互调死循环烧 Token
敏感操作	返回 pending_approval	高风险动作人工确认后再执行
幂等	写操作带 request_id	重试不产生重复工单/重复扣费

8.3 可观测性

1. 每次 tools/call 打结构化日志：tool / latency_ms / ok / error_type
2. 与前文 Actions 的失败通知打通：Server 健康检查失败 → Webhook 告警
3. 对外部依赖做熔断：连续失败则短时拒绝，避免雪崩

---

九、总结对比

对比维度	传统方式	MCP 方式
适配成本	m×n	m+n
代码维护点	各平台各一套 Plugin	一个 Server，多 Host 复用
扩展性	新工具要重写适配	新 @mcp.tool 即插即用
可治理性	依赖各宿主特性	统一在 Server 做权限 / 审计 / 限流
与 CI/CD	难统一回归	stdio 冒烟可直接挂进 Actions

落地顺序建议：

1. FastMCP 暴露 1～3 个只读工具（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 / FastMCP
• LangChain MCP 文档
• langchain-mcp-adapters
• 系列前文：GitHub Actions 深度实践：零运维搭建 CI/CD 流水线