📖 引言：从“孤岛”到“互联”

在 AI 智能体（AI Agents）爆发的前夜，我们面临着一个巨大的互操作性危机。

每一家公司、每一个开发者都在构建自己的智能体。有的智能体能查天气，有的能写代码，有的能操作数据库。但问题是，这些智能体就像一个个孤岛，彼此之间无法沟通，也无法复用。如果你想让一个 ChatGPT 智能体去访问你的 Google Drive，你需要写一套专用的插件；如果你想让 Claude 智能体做同样的事，你又得写另一套。

为了连接 N 个模型和 M 个数据源，我们需要维护 $N\times M$ 个集成接口。这是一场工程噩梦。

模型上下文协议 (MCP) 的出现，就是为了终结这场噩梦。它由 Anthropic 在 2024 年底开源，迅速成为行业事实标准。MCP 旨在成为 AI 时代的 USB 协议——通过一个标准化的接口，让任何大模型都能即插即用地连接任何数据源和工具。

本篇将带你深入 MCP 的内核，解构其客户端-服务器架构，剖析它与传统 Function Calling 的本质区别，并手把手教你使用 FastMCP 和 LangChain 构建企业级的 MCP 服务。

---

第一部分：MCP 协议的架构哲学

1.1 核心定义：AI 的 USB 接口

MCP 是一个开放标准，它规范了 AI 模型（客户端）与外部数据/工具（服务器）之间的通信方式。

• 没有 MCP 的世界：你需要为 OpenAI 写一个 WeatherPlugin，为 Gemini 写一个 WeatherTool，为 LangChain 写一个 WeatherTool。代码重复，维护困难。
• 有 MCP 的世界：你只需要写一个标准的 MCP Weather Server。OpenAI、Claude、Gemini、LangChain 都可以直接连接这个服务器，自动发现并使用其中的工具。

1.2 架构三要素：资源、提示词、工具

MCP 将外部世界的能力抽象为三个核心概念：

1. 资源 (Resources)：

   • 定义：类似于文件或数据库记录的被动数据。
   • 例子：日志文件、API 文档、数据库中的一张表。
   • 交互：客户端可以“读取”资源，也可以订阅资源的更新（实时数据流）。
   • URI：每个资源都有唯一的 URI，例如 postgres://users/schema。

2. 提示词 (Prompts)：

   • 定义：预定义的、可复用的交互模板。
   • 例子：一个“代码审查”提示词，它自动加载当前 Git 仓库的 diff 作为上下文。
   • 价值：将复杂的 Prompt Engineering 封装在服务器端，客户端只需调用 get_prompt("code_review")。

3. 工具 (Tools)：

   • 定义：可执行的函数或操作。
   • 例子：send_email、execute_sql、resize_image。
   • 交互：客户端发送参数（JSON），服务器执行逻辑并返回结果。这是最接近传统 Function Calling 的部分。

1.3 客户端-主机-服务器模型

MCP 的运行架构通常包含三个角色：

• MCP Host (主机)：这是运行 AI 模型的应用程序（如 Claude Desktop App、Cursor IDE、或你自己写的 LangChain 应用）。它负责发起连接。
• MCP Client (客户端)：Host 内部用于与 Server 通信的协议实现层（1:1 对应）。
• MCP Server (服务器)：提供数据和工具的独立进程。它可以运行在本地（Stdio 传输），也可以运行在远程（SSE/HTTP 传输）。

---

第二部分：MCP vs. 传统 Function Calling

很多开发者会问：“我已经会写 LangChain Tools 了，为什么还要学 MCP？”

特性	传统 Function Calling / Tools	MCP (Model Context Protocol)
连接方式	硬编码：工具代码必须嵌入在 Agent 代码中。	动态连接：工具运行在独立进程中，Agent 在运行时通过协议“发现”工具。
复用性	低：LangChain Tool 很难直接给 AutoGen 用。	极高：写一次 MCP Server，所有支持 MCP 的框架/IDE 都能用。
部署架构	单体：所有逻辑都在一个 Python 进程里。	分布式：数据库工具跑在数据库服务器上，文件工具跑在文件服务器上。
安全性	依赖代码审查：难以隔离。	沙箱化：Server 运行在独立进程/容器中，权限可控。
数据流	主要是“执行动作”。	包含“读取资源”、“获取提示词”和“执行动作”。

比喻：

• Function Calling 就像是你随身携带的瑞士军刀。方便，但你得一直带着它。
• MCP 就像是墙上的电源插座。你不需要随身带发电机，到了任何有插座的地方，插上就能用电。

---

第三部分：实战 FastMCP —— 极速构建服务器

FastMCP 是一个由社区驱动的 Python 框架（类似 FastAPI），专门用于简化 MCP Server 的开发。它利用 Python 的类型注解自动生成协议所需的 Schema。

3.1 环境准备

pip install fastmcp

3.2 场景一：构建一个“系统运维” MCP 服务器

假设我们需要一个 MCP 服务器，它能提供服务器的实时日志（资源），并允许执行重启服务的操作（工具）。

from fastmcp import FastMCP, Context
import subprocess
import os

# 初始化 Server
# dependencies 参数允许你声明该 Server 运行所需的依赖，FastMCP 会自动管理环境
mcp = FastMCP("DevOps Server", dependencies=["psutil"])

# --- 1. 定义资源 (Resources) ---
# 资源通常是只读的数据流
@mcp.resource("system://logs/{service_name}")
def get_service_logs(service_name: str) -> str:
    """获取指定服务的最近 50 行日志"""
    log_path = f"/var/log/{service_name}.log"
    if not os.path.exists(log_path):
        return "Log file not found."
    
    # 模拟读取日志
    try:
        # 在真实场景中，这里可能是 tail -n 50
        with open(log_path, "r") as f:
            lines = f.readlines()[-50:]
            return "".join(lines)
    except Exception as e:
        return f"Error reading logs: {str(e)}"

# --- 2. 定义工具 (Tools) ---
# 工具是可以改变系统状态的操作
@mcp.tool()
def restart_service(service_name: str, force: bool = False) -> str:
    """
    重启指定的系统服务。
    
    Args:
        service_name: 服务名称 (如 nginx, postgresql)
        force: 是否强制重启 (kill -9)
    """
    # 简单的安全检查（在真实 MCP 中，可以在 Client 端做鉴权）
    allowed_services = ["nginx", "app_worker"]
    if service_name not in allowed_services:
        return f"Error: Service '{service_name}' is not allowed to be managed via MCP."

    cmd = ["systemctl", "restart", service_name]
    if force:
        # 仅作演示，实际生产中需谨慎
        pass 
        
    try:
        # 执行命令
        result = subprocess.run(cmd, capture_output=True, text=True)
        if result.returncode == 0:
            return f"Service '{service_name}' restarted successfully."
        else:
            return f"Failed to restart: {result.stderr}"
    except Exception as e:
        return f"Execution error: {str(e)}"

# --- 3. 定义提示词 (Prompts) ---
# 预定义的交互模板，帮助 LLM 更好地使用这些工具
@mcp.prompt("diagnose_error")
def diagnose_prompt(service_name: str) -> str:
    """生成用于诊断服务错误的 Prompt"""
    return f"""
    请作为一名资深运维工程师，分析以下服务的状态。
    
    1. 首先，使用 `get_service_logs` 读取资源 `system://logs/{service_name}`。
    2. 分析日志中的 ERROR 或 WARNING 信息。
    3. 如果认为是临时故障，调用 `restart_service` 工具。
    4. 如果是配置错误，请给出修改建议。
    """

if __name__ == "__main__":
    # 启动服务器（默认使用 Stdio 模式，适合本地 CLI 集成）
    mcp.run()

3.3 运行与调试

FastMCP 提供了一个内置的 Inspector（调试器），这是开发 MCP Server 的神器。

# 在终端运行
fastmcp inspect my_server.py

这将启动一个 Web 界面，你可以在里面：

1. 查看服务器暴露的所有 Resources, Prompts 和 Tools。
2. 模拟 LLM 调用这些工具，查看返回值。
3. 查看 JSON-RPC 协议的通信日志。

---

第四部分：在 LangChain 中集成 MCP

虽然 Claude Desktop App 原生支持 MCP，但作为开发者，我们更关心如何在自己的 Python 代码（LangChain/LangGraph 应用）中连接这些 MCP Server。

在 LangChain v0.3 中，官方提供了 langchain-mcp-adapters（或者社区实现），但核心逻辑其实是通用的：将 MCP Client 包装为 LangChain Tools。

4.1 架构设计

1. MCP Client：负责与 MCP Server 建立连接（Stdio 或 SSE）。
2. 适配器 (Adapter)：
   • 调用 client.list_tools() 获取工具列表。
   • 将每个 MCP Tool 转换为 langchain_core.tools.StructuredTool。
   • 将 MCP Resources 转换为 LangChain 的 Retriever 或 DocumentLoader。

4.2 实战代码：构建通用 MCP 连接器

以下是一个基于 langchain 和 mcp 官方 SDK 的通用连接器实现。

环境准备：

pip install mcp langchain-openai langgraph

核心代码实现：

import asyncio
from contextlib import AsyncExitStack
from typing import List

from langchain_core.tools import StructuredTool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

# MCP 官方 SDK
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

class LangChainMCPClient:
    """
    一个通用的适配器，将 MCP Server 转换为 LangChain 可用的组件。
    """
    def __init__(self, command: str, args: List[str]):
        self.params = StdioServerParameters(command=command, args=args)
        self.session: ClientSession | None = None
        self.exit_stack = AsyncExitStack()

    async def connect(self):
        """建立与 MCP Server 的连接"""
        # 启动子进程并建立通信通道
        read, write = await self.exit_stack.enter_async_context(stdio_client(self.params))
        self.session = await self.exit_stack.enter_async_context(
            ClientSession(read, write)
        )
        await self.session.initialize()
        
        # 发现工具
        tools_list = await self.session.list_tools()
        print(f"🔗 Connected to MCP Server. Found {len(tools_list.tools)} tools.")
        return tools_list

    async def get_langchain_tools(self) -> List[StructuredTool]:
        """将 MCP Tools 转换为 LangChain Tools"""
        if not self.session:
            await self.connect()
            
        mcp_tools = (await self.session.list_tools()).tools
        langchain_tools = []

        for tool in mcp_tools:
            # 闭包捕获 tool.name
            async def _tool_wrapper(
                tool_name=tool.name, 
                **kwargs
            ):
                # 实际调用 MCP Server
                result = await self.session.call_tool(tool_name, arguments=kwargs)
                # MCP 返回的是 Content 列表，通常我们要提取文本
                return result.content[0].text

            # 构建 LangChain Tool
            lc_tool = StructuredTool.from_function(
                func=None,
                coroutine=_tool_wrapper, # 使用异步调用
                name=tool.name,
                description=tool.description,
                # MCP 的 inputSchema 是 JSON Schema，LangChain 可以直接用（或稍作转换）
                # 这里为了简化，我们依赖 LangChain 的自动推断，或者你需要手动转换 Pydantic
            )
            langchain_tools.append(lc_tool)
            
        return langchain_tools

    async def cleanup(self):
        await self.exit_stack.aclose()

# --- 业务逻辑 ---

async def main():
    # 假设我们有一个运行 SQLite 的 MCP Server
    # (你可以使用官方的 @modelcontextprotocol/server-sqlite)
    # 这里我们连接到一个本地运行的 Python MCP Server
    client = LangChainMCPClient(
        command="python", 
        args=["my_devops_server.py"] # 上一节写的 Server
    )

    try:
        # 1. 获取工具
        tools = await client.get_langchain_tools()
        
        # 2. 初始化 LLM
        llm = ChatOpenAI(model="gpt-4o", temperature=0)
        
        # 3. 构建 LangGraph Agent (ReAct 模式)
        agent_executor = create_react_agent(llm, tools)
        
        # 4. 执行任务
        print("\n🤖 Agent 启动...")
        query = "请检查 nginx 服务的日志，如果发现错误，请帮我重启它。"
        
        # Stream 模式查看执行过程
        async for event in agent_executor.astream(
            {"messages": [("user", query)]}
        ):
            for value in event.values():
                print("--- Step ---")
                print(value["messages"][-1].content)
                
    finally:
        await client.cleanup()

if __name__ == "__main__":
    asyncio.run(main())

代码解析：

1. stdio_client：这是 MCP 的核心传输层。它启动子进程（你的 Python Server），并通过标准输入输出（Stdin/Stdout）进行 JSON-RPC 通信。
2. list_tools：客户端向服务器发送握手请求，获取工具清单（Schema）。
3. 适配层：我们动态创建了 Python 函数 _tool_wrapper，它内部调用 session.call_tool。这样 LangChain 就会认为这只是一个普通的 Async Tool。
4. create_react_agent：LangGraph 的预构建 Agent，它能够理解 Tools 的描述，规划调用顺序（先查日志 -> 分析 -> 重启），这正是第十篇的核心应用。

---

第五部分：Google ADK 的 MCP 原生支持

Google 的 ADK 对 MCP 的支持更加原生和开箱即用。它提供了一个 MCPToolset 类，封装了上述的所有连接逻辑。

5.1 实战：ADK 连接文件系统 MCP

假设你想让 Gemini 智能体能够读写你电脑上的文件。你可以使用官方的 filesystem MCP Server（Node.js 实现）。

import os
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters

# 1. 准备目标目录
TARGET_DIR = os.path.abspath("./workspace")
os.makedirs(TARGET_DIR, exist_ok=True)

# 2. 定义智能体
agent = LlmAgent(
    model='gemini-2.0-flash',
    name='FileAssistant',
    instruction=f'你是一个文件管理助手。你可以读写 {TARGET_DIR} 目录下的文件。',
    # 3. 注入 MCP 工具集
    tools=[
        MCPToolset(
            connection_params=StdioServerParameters(
                # 使用 npx 运行社区提供的 MCP Server
                command='npx',
                args=[
                    "-y", 
                    "@modelcontextprotocol/server-filesystem", 
                    TARGET_DIR
                ],
            ),
            # 可选：权限控制，只暴露读接口，隐藏写接口
            # tool_filter=['list_directory', 'read_file']
        )
    ]
)

# 注意：ADK 会自动处理 Server 的启动、保活和关闭

5.2 进阶：远程 HTTP 连接 (SSE)

除了本地进程（Stdio），MCP 还支持远程连接（Server-Sent Events, SSE）。这对于分布式系统非常有用。

from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, HttpServerParameters

remote_tools = MCPToolset(
    connection_params=HttpServerParameters(
        url="http://mcp-database-service.internal:8000/sse"
    )
)

---

第六部分：企业级 MCP 架构设计模式

在企业内部落地 MCP 时，不仅仅是写几个 Tool 那么简单。我们需要考虑整体架构。

6.1 模式一：Sidecar 模式 (Local Stdio)

• 架构：每个 Agent 实例启动时，作为子进程拉起所需的 MCP Server。
• 适用：本地文件操作、Git 操作、单机数据分析。
• 优点：低延迟，安全（Server 只对当前 Agent 可见）。
• 缺点：资源消耗大（每个 Agent 都要起一套 Server），无法共享状态。

6.2 模式二：Gateway 模式 (Remote HTTP/SSE)

• 架构：部署一个中心化的 MCP Gateway 集群。所有 Agent 通过 HTTP 连接到 Gateway。
• 适用：企业知识库（RAG）、核心数据库操作、昂贵的 API 代理。
• 优点：资源共享（连接池复用）、统一鉴权、日志审计。
• 缺点：网络延迟，增加了中心化故障点。

6.3 模式三：Agent as a Server (递归 MCP)

这是一个非常有趣的高级模式。

• 概念：一个构建好的 Agent 本身，也可以通过 MCP 协议暴露出去，变成另一个 Agent 的“工具”。
• 场景：
  • 你构建了一个复杂的“法律顾问 Agent”（包含 RAG、推理、反思）。
  • 你可以用 FastMCP 把它包装成一个 Server。
  • 外部的“总助 Agent”连接这个 Server，把它当作一个 consult_lawyer() 工具来调用。
• 实现：FastMCP 支持直接将 LangGraph 应用注册为 Tool。

# 伪代码示例
@mcp.tool()
async def consult_legal_department(query: str) -> str:
    """咨询法律部门（这是一个子智能体）"""
    # 内部调用 LangGraph Agent
    response = await legal_agent.ainvoke({"messages": [query]})
    return response["messages"][-1].content

---

结语：互联互通的 AI 生态

模型上下文协议 (MCP) 不仅仅是一个技术标准，它是一种生态宣言。

• 对于工具开发者（如 Linear, Notion, Stripe）：只需维护一套 MCP Server，就能接入所有 AI 模型。
• 对于模型开发者（如 Anthropic, Google）：只需支持 MCP Client，就能立刻拥有海量的外部能力。
• 对于应用开发者（我们）：可以像搭积木一样，自由组合最好的模型和最好的数据源。

随着 MCP 生态的成熟，未来的 AI 应用开发将不再是编写一个个孤立的 Bot，而是编排一个由无数 MCP Server 组成的庞大协作网络。

这就是 AI 的“万物互联”时代。

参考资料

1.Model Context Protocol (MCP) Documentation. (Latest). Model Context Protocol (MCP). https://google.github.io/adk-docs/mcp/
2.FastMCP Documentation. FastMCP. https://github.com/jlowin/fastmcp
3.MCP Tools for Genmedia Services. MCP Tools for Genmedia Services. https://google.github.io/adk-docs/mcp/#mcp-servers-for-google-cloud-genmedia
4.MCP Toolbox for Databases Documentation. (Latest). MCP Toolbox for Databases. https://google.github.io/adk-docs/mcp/databases/