导读：还在为每个新数据源写自定义 Tool？还在纠结如何安全地让 AI 访问本地文件、数据库或内部 API？MCP (Model Context Protocol) 来了！这是继 Function Calling 之后，大模型生态最重大的基础设施升级。

本文将带你深入 MCP 架构，通过 LangChain + MCP 的实战组合，构建一个能无缝读取本地 SQLite 数据库、实时监控文件系统、调用内部 REST API 的超级 Agent。无需重复造轮子，只需“插拔”即可赋予 AI 无限上下文！

---

🌐 一、为什么需要 MCP？痛点与变革

在 MCP 出现之前，给 LLM 添加能力（Tools）通常面临三大痛点：

1. 开发成本高：每接入一个新数据源（如 Notion、Slack、MySQL），都要手写一套 Tool 类，处理鉴权、分页、错误重试。
2. 安全边界模糊：直接在代码中硬编码数据库密码或 API Key，容易泄露；且难以精细控制 AI 的读写权限。
3. 生态割裂：A 公司的 Agent 无法直接复用 B 公司开发的数据连接器。

💡 MCP 的核心愿景

MCP (Model Context Protocol) 由 Anthropic 牵头提出，旨在定义一种通用标准协议。

• 对于开发者：你只需要写一次 MCP Server（数据源适配器），所有支持 MCP 的客户端（Claude Desktop, LangChain, LlamaIndex 等）都能直接连接使用。
• 对于 Agent：它不再需要知道数据存在哪里，只需通过标准协议请求“资源 (Resources)”、“提示词 (Prompts)”或“工具 (Tools)”。

🔄 MCP 架构工作流

---

🛠️ 二、环境准备：搭建 MCP 生态

要运行本实战，我们需要三个部分：

1. MCP Server：提供数据能力的服务端（我们将使用 Python 快速构建一个）。
2. MCP Client：LangChain 中的连接组件。
3. LLM：支持复杂推理的大模型。

1. 安装依赖

# 核心库
pip install langchain langchain-openai mcp uvicorn

# MCP Python SDK (用于编写 Server)
pip install mcp-python-sdk

# 示例数据源依赖
pip install aiosqlite pandas

---

💻 三、实战第一步：构建自定义 MCP Server

假设我们有一个本地的 analytics.db (SQLite)，里面存着用户订单数据。我们要创建一个 MCP Server，让 AI 能安全查询它，而无需暴露原始 SQL 执行权限。

创建文件 my_data_server.py：

import asyncio
import sqlite3
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Resource, Tool, TextContent

# 初始化 Server
app = Server("local-data-analyzer")

@app.list_resources()
async def list_resources() -> list[Resource]:
    """告诉 Client 我们有哪些数据资源"""
    return [
        Resource(
            uri="sqlite://orders",
            name="用户订单表",
            description="包含最近一年的电商订单数据 (order_id, user_id, amount, date)",
            mimeType="application/json",
        )
    ]

@app.read_resource()
async def read_resource(uri: str) -> str:
    """读取资源的具体内容"""
    if uri == "sqlite://orders":
        conn = sqlite3.connect("analytics.db")
        cursor = conn.cursor()
        # 安全起见，只允许读取前 10 条作为示例，实际可配合 Tool 进行复杂查询
        cursor.execute("SELECT * FROM orders LIMIT 10")
        rows = cursor.fetchall()
        conn.close()
        return str(rows)
    raise ValueError(f"Unknown resource: {uri}")

@app.list_tools()
async def list_tools() -> list[Tool]:
    """暴露给 AI 的工具：查询特定用户的订单总额"""
    return [
        Tool(
            name="query_user_total",
            description="查询指定 user_id 的历史订单总金额",
            inputSchema={
                "type": "object",
                "properties": {
                    "user_id": {"type": "integer", "description": "用户唯一标识 ID"}
                },
                "required": ["user_id"],
            },
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """执行工具逻辑"""
    if name == "query_user_total":
        user_id = arguments["user_id"]
        conn = sqlite3.connect("analytics.db")
        cursor = conn.cursor()
        cursor.execute("SELECT SUM(amount) FROM orders WHERE user_id = ?", (user_id,))
        result = cursor.fetchone()[0]
        conn.close()
        return [TextContent(type="text", text=f"用户 {user_id} 的总消费为: ${result}")]
    
    raise ValueError(f"Unknown tool: {name}")

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_session())

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

💡 亮点：这个 Server 完全独立于 LangChain。你可以用 Node.js 写，也可以用 Rust 写，只要遵循 MCP 协议，LangChain 都能连！

---

🔗 四、实战第二步：LangChain 集成 MCP Client

现在，我们在 LangChain 中启动这个 Server，并将其转换为标准的 LangChain Tools。

创建文件 agent_with_mcp.py：

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

# 1. 配置 MCP 客户端
# 这里我们通过 stdio (标准输入输出) 启动刚才写的 python 脚本
# 在生产环境中，也可以连接远程的 SSE 端点
mcp_client = MultiServerMCPClient(
    {
        "local_db": {
            "command": "python",
            "args": ["my_data_server.py"], # 启动我们的 MCP Server
            "transport": "stdio"
        }
    }
)

# 2. 异步获取 Tools
# 这一步会将 MCP Server 暴露的 'query_user_total' 转换为 LangChain 的 BaseTool 对象
async def get_tools():
    tools = await mcp_client.get_tools()
    return tools

# 3. 初始化 LLM
llm = ChatOpenAI(model="gpt-4-turbo", temperature=0)

# 4. 构建 Agent
async def run_agent():
    tools = await get_tools()
    
    # 打印一下加载到的工具，确认成功
    print(f"✅ 成功加载 MCP 工具: {[t.name for t in tools]}")

    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个数据分析助手。你可以访问本地数据库工具。请根据用户请求使用工具回答问题。"),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}")
    ])

    agent = create_react_agent(llm, tools, prompt)
    executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

    # 5. 测试提问
    query = "帮我查一下用户 ID 为 1001 的总消费金额是多少？"
    print(f"\n🤖 用户问: {query}")
    result = executor.invoke({"input": query})
    print(f"\n💡 回答: {result['output']}")

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

🖥️ 运行效果预期

✅ 成功加载 MCP 工具: ['query_user_total']

🤖 用户问：帮我查一下用户 ID 为 1001 的总消费金额是多少？

> Entering new AgentExecutor chain...
Thought: 用户想查询特定用户的总消费，我需要使用 'query_user_total' 工具。
Action: query_user_total
Action Input: {"user_id": 1001}
Observation: 用户 1001 的总消费为: $2580.50
Thought: 我已经获取到了数据，可以直接回答。
Final Answer: 用户 ID 1001 的历史订单总消费金额为 $2580.50。

> Finished chain.

💡 回答: 用户 ID 1001 的历史订单总消费金额为 $2580.50。

---

🌟 五、进阶场景：多服务器协作与动态资源

MCP 的强大之处在于组合。你可以同时连接多个 Server：

1. File System Server：让 AI 读取本地项目代码库。
2. Postgres Server：连接生产环境只读库。
3. Slack Server：读取团队聊天记录。

架构图：多源融合

代码修改示意：
只需在 MultiServerMCPClient 配置字典中添加新的 entry：

mcp_client = MultiServerMCPClient({
    "local_db": {"command": "python", "args": ["my_data_server.py"]},
    "file_system": {"command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "/Users/dev/project"]},
    "postgres": {"url": "http://localhost:8080/sse"} # 远程 SSE 服务
})

注：@modelcontextprotocol/server-filesystem 是社区提供的现成文件服务器，无需自己写代码！

---

🛡️ 六、安全性与企业级考量

在企业落地 MCP 时，必须关注以下安全问题：

1. 最小权限原则：
   • MCP Server 应以低权限用户运行。
   • 在 call_tool 中进行二次校验（如：禁止 DELETE 操作，限制 SELECT 行数）。
2. 传输加密：
   • 本地 stdio 是安全的。
   • 跨网络通信务必使用 SSE over HTTPS 或 WebSocket Secure (WSS)。
3. 人类在环 (Human-in-the-loop)：
   • 对于高风险操作（如写入数据库、发送邮件），在 LangChain 层增加 approval 步骤，要求用户确认后再调用 MCP Tool。

---

📝 七、总结：MCP 是 Agent 的“USB-C 接口”

如果说 LLM 是电脑的大脑，那么 MCP 就是通用的 USB-C 接口。

• 过去：每个外设（数据源）都需要定制驱动（手写 Tool）。
• 现在：只要设备支持 MCP 协议，插上（配置 Server）就能用。

通过本次实战，我们实现了：

1. ✅ 解耦：数据逻辑与 Agent 逻辑彻底分离。
2. ✅ 复用：一个 MCP Server 可被 Claude、LangChain、LlamaIndex 同时使用。
3. ✅ 扩展：轻松组合文件、数据库、API 等多源数据。

🚀 下一步行动：
不要再去写重复的 Connector 了！尝试为你公司的内部系统（如 Jira, ERP, 日志系统）编写一个 MCP Server，让你的 AI 助手瞬间获得“企业级视力”！