一、简介

LangChain 模型上下文协议（MCP，Model Context Protocol）是一套标准化、跨模型、可插拔的协议，用于让 LLM/Agent 统一发现、调用外部工具、资源与提示模板，解决工具集成碎片化、跨模型适配成本高、安全与权限难管控的问题LangChain。它让 LangChain Agent 能像 “插 USB” 一样接入各类外部能力，是企业级 Agent 架构的关键基础设施。

1.1 解决的核心问题

在传统的AI应用中，集成外部工具面临以下问题：

1. 工具集成碎片化：不同模型（GPT-4、Claude、通义千问）、不同框架（LangChain、AutoGPT）的工具调用格式不统一，适配成本高。
2. 动态扩展难：新增工具需修改 Agent 代码、重启服务，无法 “热插拔”。
3. 安全与隔离弱：工具执行与 LLM 应用同进程，权限、审计、限流难管控。
4. 上下文共享低效：RAG、记忆、工具返回的上下文无统一格式，难以跨组件复用。

1.2 核心价值

1. 统一接口：LLM 与外部能力的 “USB 协议”，一次适配、全模型兼容。
2. 动态发现：Agent 可自动发现 MCP Server 提供的工具 / 资源，无需硬编码。
3. 安全隔离：工具在独立 MCP Server 执行，支持权限、审计、限流。
4. 生态互通：第三方可发布 MCP 兼容服务（如 SQL、文件、API），形成 “AI 插件市场”。

1.3 MCP vs 其他协议

协议	定位	特点
MCP	AI-工具交互	专注AI上下文，支持工具发现、资源访问
OpenAPI	RESTful API	传统HTTP API标准，非AI专用
Function Calling	模型调用	模型厂商特有，不统一
Agent Protocol	智能体通信	智能体间通信，非工具集成

二、MCP 核心架构与组件

2.1 架构图

┌─────────────────────────────────────────────────────────────┐
│                    AI应用 (LangChain)                       │
│  ┌─────────────────────────────────────────────────────┐    │
│  │              MCP Client (客户端)                    │    │
│  │  ┌─────────────┐ ┌─────────────┐ ┌─────────────┐   │    │
│  │  │ Tool 调用   │ │ Resource    │ │ Prompt      │   │    │
│  │  │ Handler     │ │ Access      │ │ Templates   │   │    │
│  │  └─────────────┘ └─────────────┘ └─────────────┘   │    │
│  └────────────────────────────────────────────────────┘    │
│                            │                               │
│                            │ MCP Protocol                  │
│                            ▼                               │
│  ┌───────────────────────────────────────────────────┐     │
│  │              MCP Server (服务器)                  │     │
│  │  ┌────────────────────────────────────────────┐   │    │
│  │  │          Tool 实现 / Resource 提供          │  │     │
│  │  └────────────────────────────────────────────┘  │     │
│  └──────────────────────────────────────────────────┘     │
│                            │                              │
│        ┌───────────────────┼───────────────────┐          │
│        ▼                   ▼                   ▼          │
│   ┌─────────┐       ┌─────────┐        ┌─────────┐        │
│   │Database │       │   API   │        │  Files  │        │
│   └─────────┘       └─────────┘        └─────────┘        │
└───────────────────────────────────────────────────────────┘

2.2 核心概念

概念	描述	类比理解
MCP Client	运行在AI应用中的客户端，负责与MCP服务器通信	浏览器（访问网站）
MCP Server	提供工具、资源、提示模板的服务器	网站服务器（提供服务）
Tool	可执行的函数，AI可以调用	API接口
Resource	可访问的数据资源（文件、数据库等）	文件系统
Prompt Template	预定义的提示词模板	表单模板
Transport	通信方式（stdio、SSE、WebSocket）	网络协议

2.3 三大核心能力

MCP 定义 Server 可提供三类标准化能力，是 Agent 与外部交互的核心接口：

1. Tools（工具）

   • 定义：可执行函数，Agent 可调用完成操作（数据库查询、API 调用、文件读写）。
   • LangChain 映射：MCP Tools → LangChain Tools，直接用于 Agent 工具调用。
   • 关键能力：
     • 自动发现：client.get_tools() 获取所有可用工具。
     • 结构化返回：支持返回机器可读数据（JSON）+ 人类可读文本LangChain。
     • 权限控制：Server 侧做权限校验、限流、审计。

2. Resources（资源）

   • 定义：上下文数据源（文件内容、数据库记录、API 响应、知识库片段）。
   • 核心作用：为 RAG、Agent 提供标准化上下文，支持按 URI 访问、分页、过滤。
   • 示例：file:///docs/report.pdf、db://sales/orders、api://weather/current。

3. Prompts（提示模板）

   • 定义：可重用提示模板（系统提示、少样本示例、任务指令）。
   • 价值：统一管理提示词，支持动态加载、版本控制、跨应用共享。

2.4 MCP 传输方式（Client ↔ Server）

支持三种传输，适配本地 / 远程、实时 / 非实时场景：

1. stdio（标准输入输出）

   • 适用：本地工具、简单场景。
   • 原理：Client 启动 Server 作为子进程，通过 stdin/stdout 通信。
   • 优势：部署简单、无网络依赖。
   • 示例：本地 Python MCP Server、文件系统工具。

2. Streamable HTTP（可流式 HTTP）

   • 适用：远程 Server、多客户端、生产环境。
   • 原理：Server 作为独立 HTTP 服务，Client 发送 POST 请求，支持长连接、流式响应。
   • 优势：跨网络、可负载均衡、支持多客户端。
   • 示例：云部署的 SQL Server、API 网关。

3. Server-Sent Events (SSE)

   • 适用：实时场景（日志推送、状态更新）。
   • 原理：基于 HTTP 的单向流式通信，Server 主动推送数据。
   • 优势：低延迟、实时性强。
   • 示例：Agent 执行进度、实时监控。

2.5 MCP 消息类型

# 1. 请求消息
{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/call",
    "params": {
        "name": "search_web",
        "arguments": {"query": "AI news"}
    }
}

# 2. 响应消息
{
    "jsonrpc": "2.0",
    "id": 1,
    "result": {
        "content": [{"type": "text", "text": "搜索结果..."}]
    }
}

# 3. 通知消息（无需响应）
{
    "jsonrpc": "2.0",
    "method": "notifications/initialized"
}

三、LangChain中的MCP客户端

LangChain提供了对MCP的原生支持，通过langchain-mcp包实现。

3.1 安装

pip install langchain-mcp
# 或完整安装
pip install langchain-mcp[all]

3.2 基础MCP客户端

from langchain_mcp import MCPClient
from langchain_mcp.tools import MCPTool

# 1. 创建MCP客户端
client = MCPClient(
    server_command=["python", "my_mcp_server.py"],  # 启动服务器的命令
    transport="stdio",  # 通信方式
    timeout=30  # 超时时间
)

# 2. 连接服务器
await client.connect()

# 3. 获取可用工具
tools = await client.list_tools()
print(f"可用工具: {[t.name for t in tools]}")

# 4. 调用工具
result = await client.call_tool(
    name="search_web",
    arguments={"query": "LangChain MCP"}
)

# 5. 获取资源
resources = await client.list_resources()
resource_content = await client.read_resource("file:///data/config.json")

# 6. 关闭连接
await client.close()

3.3 同步版本

from langchain_mcp import MCPClientSync

client = MCPClientSync(
    server_command=["python", "my_mcp_server.py"]
)

client.connect()
tools = client.list_tools()
result = client.call_tool("search_web", {"query": "AI"})
client.close()

3.4 集成LangChain工具

from langchain_mcp import MCPTool
from langchain.agents import create_agent

# 1. 从MCP服务器创建LangChain工具
mcp_client = MCPClientSync(server_command=["python", "server.py"])
mcp_client.connect()

langchain_tools = []
for tool_info in mcp_client.list_tools():
    # 将每个MCP工具包装为LangChain工具
    tool = MCPTool(
        name=tool_info.name,
        description=tool_info.description,
        client=mcp_client,
        input_schema=tool_info.inputSchema
    )
    langchain_tools.append(tool)

# 2. 创建智能体
agent = create_agent(
    model="openai:gpt-4o",
    tools=langchain_tools,
    system_prompt="你可以使用各种工具帮助用户。"
)

# 3. 使用
result = agent.invoke({
    "messages": [{"role": "user", "content": "搜索最新的AI新闻"}]
})

四、构建MCP服务器

4.1 基础MCP服务器

# my_mcp_server.py
import asyncio
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.server.stdio
import mcp.types as types

# 创建服务器实例
server = Server("my-mcp-server")

# 定义工具
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    """列出所有可用工具"""
    return [
        types.Tool(
            name="search_web",
            description="搜索网络信息",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "搜索关键词"
                    }
                },
                "required": ["query"]
            }
        ),
        types.Tool(
            name="calculator",
            description="数学计算",
            inputSchema={
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "数学表达式，如 '2+3*4'"
                    }
                },
                "required": ["expression"]
            }
        )
    ]

# 实现工具调用
@server.call_tool()
async def handle_call_tool(
    name: str, 
    arguments: dict | None
) -> list[types.TextContent]:
    """处理工具调用"""
    
    if name == "search_web":
        query = arguments.get("query", "")
        # 模拟搜索
        result = f"搜索结果：关于 '{query}' 的信息..."
        return [types.TextContent(type="text", text=result)]
    
    elif name == "calculator":
        expression = arguments.get("expression", "")
        try:
            # 安全计算（生产环境使用更安全的eval替代方案）
            result = eval(expression)
            return [types.TextContent(
                type="text", 
                text=f"{expression} = {result}"
            )]
        except Exception as e:
            return [types.TextContent(
                type="text",
                text=f"计算错误: {str(e)}"
            )]
    
    raise ValueError(f"Unknown tool: {name}")

# 定义资源
@server.list_resources()
async def handle_list_resources() -> list[types.Resource]:
    """列出可用资源"""
    return [
        types.Resource(
            uri="file:///data/config.json",
            name="配置文件",
            description="应用配置文件",
            mimeType="application/json"
        )
    ]

@server.read_resource()
async def handle_read_resource(uri: str) -> str:
    """读取资源"""
    if uri == "file:///data/config.json":
        return '{"version": "1.0", "name": "my-app"}'
    raise ValueError(f"Resource not found: {uri}")

# 启动服务器
async def main():
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="my-mcp-server",
                server_version="1.0.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={}
                )
            )
        )

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

4.2 高级MCP服务器（带资源模板）

import json
import os
from typing import Any
from mcp.server import Server
from mcp.types import ResourceTemplate

server = Server("advanced-mcp-server")

# 动态资源模板
@server.list_resource_templates()
async def handle_list_resource_templates() -> list[ResourceTemplate]:
    """列出资源模板"""
    return [
        ResourceTemplate(
            uri_template="file:///{path}",
            name="文件系统",
            description="访问本地文件"
        ),
        ResourceTemplate(
            uri_template="db:///{table}/{id}",
            name="数据库记录",
            description="访问数据库记录"
        )
    ]

@server.read_resource()
async def handle_read_resource_with_template(uri: str) -> str:
    """根据URI模板读取资源"""
    
    if uri.startswith("file:///"):
        # 提取文件路径
        path = uri[7:]  # 去掉 "file:///"
        if os.path.exists(path):
            with open(path, 'r') as f:
                return f.read()
        else:
            raise ValueError(f"File not found: {path}")
    
    elif uri.startswith("db:///"):
        # 解析数据库URI
        parts = uri[5:].split('/')  # 去掉 "db://"
        if len(parts) >= 2:
            table = parts[0]
            record_id = parts[1]
            # 模拟数据库查询
            return json.dumps({
                "table": table,
                "id": record_id,
                "data": "模拟数据"
            })
    
    raise ValueError(f"Unsupported URI: {uri}")

# 带进度的工具调用
@server.call_tool()
async def handle_call_tool_with_progress(
    name: str, 
    arguments: dict | None
) -> list[types.TextContent]:
    
    if name == "long_running_task":
        # 模拟长时间任务
        for i in range(10):
            # 发送进度通知
            await server.request_context.session.send_notification(
                "notifications/progress",
                {
                    "progress": i * 10,
                    "total": 100,
                    "message": f"处理中... {i*10}%"
                }
            )
            await asyncio.sleep(0.5)
        
        return [types.TextContent(
            type="text",
            text="任务完成！"
        )]
    
    raise ValueError(f"Unknown tool: {name}")

4.3 使用SSE传输

# sse_server.py
from mcp.server import Server
import mcp.server.sse

server = Server("sse-mcp-server")

# ... 定义工具和资源 ...

# 启动SSE服务器
async def main():
    import uvicorn
    from mcp.server.sse import SseServerTransport
    
    transport = SseServerTransport("/messages")
    
    app = transport.create_app(server)
    
    config = uvicorn.Config(app, host="127.0.0.1", port=8000)
    server_runner = uvicorn.Server(config)
    await server_runner.serve()

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

客户端连接SSE服务器：

from langchain_mcp import MCPClient

client = MCPClient(
    server_url="http://localhost:8000/sse",
    transport="sse"
)

await client.connect()
tools = await client.list_tools()

五、MCP内置服务器

LangChain和MCP社区提供了多种内置服务器，开箱即用。

5.1 文件系统服务器

# 启动文件系统服务器
from mcp_server_filesystem import create_filesystem_server

server = create_filesystem_server(
    allowed_directories=["/data", "/home/user/docs"]
)

# 客户端使用
client = MCPClient(server_command=["python", "-m", "mcp_server_filesystem"])
client.connect()

# 读取文件
content = await client.read_resource("file:///data/config.json")

# 列出目录
resources = await client.list_resources()

5.2 数据库服务器

# PostgreSQL服务器
from mcp_server_postgres import create_postgres_server

server = create_postgres_server(
    connection_string="postgresql://user:pass@localhost/db"
)

# 客户端使用
client = MCPClient(server_command=["python", "-m", "mcp_server_postgres"])
client.connect()

# 执行SQL查询
result = await client.call_tool(
    "execute_sql",
    {"query": "SELECT * FROM users LIMIT 10"}
)

5.3 浏览器自动化服务器

# Puppeteer服务器（浏览器自动化）
client = MCPClient(
    server_command=["npx", "@modelcontextprotocol/server-puppeteer"]
)
client.connect()

# 截图
result = await client.call_tool(
    "screenshot",
    {"url": "https://example.com"}
)

# 执行JavaScript
result = await client.call_tool(
    "evaluate",
    {"script": "document.title"}
)

5.4 常用内置服务器

服务器	功能	启动命令
filesystem	文件系统访问	mcp-server-filesystem
postgres	PostgreSQL查询	mcp-server-postgres
puppeteer	浏览器自动化	npx @modelcontextprotocol/server-puppeteer
fetch	HTTP请求	mcp-server-fetch
memory	向量记忆	mcp-server-memory
sqlite	SQLite数据库	mcp-server-sqlite
github	GitHub操作	mcp-server-github
slack	Slack集成	mcp-server-slack

六、LangChain 集成 MCP 实战

步骤 1：环境准备

# 安装 MCP 适配器
pip install langchain-mcp-adapters langchain-openai

步骤 2：编写 MCP Server（提供工具）

创建 math_server.py（本地 MCP Server，提供加法 / 乘法工具）：

from mcp.server import Server
from mcp.types import Tool, TextContent

# 初始化 MCP Server
server = Server("math-server")

# 定义加法工具
@server.tool(
    name="add",
    description="计算两个数的和",
    inputSchema={
        "type": "object",
        "properties": {
            "a": {"type": "number", "description": "第一个数"},
            "b": {"type": "number", "description": "第二个数"}
        },
        "required": ["a", "b"]
    }
)
async def add(a: float, b: float) -> list[TextContent]:
    return [TextContent(type="text", text=f"结果：{a + b}")]

# 定义乘法工具
@server.tool(
    name="multiply",
    description="计算两个数的积",
    inputSchema={
        "type": "object",
        "properties": {
            "a": {"type": "number", "description": "第一个数"},
            "b": {"type": "number", "description": "第二个数"}
        },
        "required": ["a", "b"]
    }
)
async def multiply(a: float, b: float) -> list[TextContent]:
    return [TextContent(type="text", text=f"结果：{a * b}")]

# 启动 Server
if __name__ == "__main__":
    import asyncio
    asyncio.run(server.run())

步骤 3：LangChain Agent 调用 MCP 工具

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

# 1. 初始化 MCP Client（连接本地 math Server）
async def main():
    client = MultiServerMCPClient(
        {
            "math": {
                "transport": "stdio",  # 本地子进程通信
                "command": "python",
                "args": ["math_server.py"]  # MCP Server 路径
            }
        }
    )

    # 2. 从 MCP Server 发现并获取工具
    tools = await client.get_tools()
    print(f"发现 MCP 工具：{[tool.name for tool in tools]}")

    # 3. 初始化 LangChain Agent
    llm = ChatOpenAI(model="gpt-3.5-turbo", api_key="YOUR_OPENAI_KEY")
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是数学助手，使用提供的工具计算问题。"),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}")
    ])

    # 4. 创建 Agent 并绑定 MCP 工具
    agent = create_openai_tools_agent(llm, tools, prompt)
    agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

    # 5. 调用 Agent（自动使用 MCP 工具）
    response = await agent_executor.ainvoke({"input": "3加5等于多少？再乘以2"})
    print(f"最终回答：{response['output']}")

    # 关闭 MCP Client
    await client.close()

# 运行
asyncio.run(main())

步骤 4：连接远程 MCP Server（Streamable HTTP）

# 连接远程 HTTP MCP Server（如 http://localhost:8000/mcp）
client = MultiServerMCPClient(
    {
        "weather": {
            "transport": "streamable_http",
            "url": "http://localhost:8000/mcp"  # 远程 MCP Server 地址
        }
    }
)

七、MCP 生命周期与能力协商

7.1 生命周期（有状态）

1. 初始化（Initialize）：Client 发送初始化请求，Server 返回支持的能力（Tools/Resources/Prompts）。
2. 能力协商（Capabilities）：双方确认支持的原语、传输、版本。
3. 运行时（Runtime）：Client 调用工具 / 获取资源 / 加载提示；Server 执行并返回结果。
4. 关闭（Shutdown）：Client 发送关闭通知，Server 清理资源。

7.2 能力发现（核心机制）

• 自动发现：client.get_tools()、client.list_resources()、client.list_prompts() 自动获取 Server 能力。
• 无需硬编码：Agent 可动态适配 Server 新增 / 删除的工具，无需修改代码。

八、MCP 高级特性

8.1 资源访问（Resources）

标准化访问外部数据：

# 获取 MCP 资源（如文件内容）
resource = await client.read_resource("file:///docs/report.pdf")
print(f"资源内容：{resource.content}")

8.2 提示模板（Prompts）

统一管理提示词：

# 加载 MCP 提示模板
prompt_template = await client.get_prompt("system-prompt")
print(f"提示模板：{prompt_template.content}")

8.3 安全与管控

• 权限控制：Server 侧实现 RBAC，限制工具 / 资源访问。
• 审计日志：记录所有工具调用、资源访问，便于追溯。
• 限流熔断：防止恶意调用、保护后端服务。

from mcp.server import Server
import asyncio

class SecureMCPServer:
    """带权限控制的MCP服务器"""
    
    def __init__(self):
        self.server = Server("secure-mcp-server")
        self.permissions = {
            "read_file": ["user1", "admin"],
            "write_file": ["admin"],
            "execute_sql": ["admin"],
            "search_web": ["user1", "user2", "admin"]
        }
        self.user = None
    
    def set_user(self, user: str):
        self.user = user
    
    def check_permission(self, tool_name: str) -> bool:
        """检查权限"""
        allowed_users = self.permissions.get(tool_name, [])
        return self.user in allowed_users
    
    async def run(self):
        # 包装工具调用，添加权限检查
        original_call_tool = self.server.call_tool
        
        @self.server.call_tool()
        async def call_tool_with_auth(
            name: str, 
            arguments: dict | None
        ):
            if not self.check_permission(name):
                return [types.TextContent(
                    type="text",
                    text=f"权限不足：用户 {self.user} 无法调用 {name}"
                )]
            
            # 调用原始实现
            return await original_call_tool(name, arguments)
        
        # 启动服务器
        async with mcp.server.stdio.stdio_server() as (read, write):
            await self.server.run(read, write, ...)

8.4 多 Server 管理

同时连接多个 MCP Server（如 math、weather、database），Agent 自动选择合适工具。

from langchain_mcp import MultiServerClient

class MCPManager:
    """管理多个MCP服务器"""
    
    def __init__(self):
        self.clients = {}
        self.all_tools = []
    
    async def add_server(self, name: str, server_config: dict):
        """添加MCP服务器"""
        from langchain_mcp import MCPClient
        
        client = MCPClient(
            server_command=server_config.get("command"),
            server_url=server_config.get("url"),
            transport=server_config.get("transport", "stdio")
        )
        
        await client.connect()
        self.clients[name] = client
        
        # 收集工具
        tools = await client.list_tools()
        for tool in tools:
            self.all_tools.append({
                "server": name,
                "tool": tool,
                "client": client
            })
    
    async def call_tool(self, server_name: str, tool_name: str, arguments: dict):
        """调用指定服务器的工具"""
        client = self.clients.get(server_name)
        if not client:
            raise ValueError(f"Server {server_name} not found")
        
        return await client.call_tool(tool_name, arguments)
    
    async def search_tools(self, keyword: str):
        """搜索工具"""
        results = []
        for tool_info in self.all_tools:
            if keyword.lower() in tool_info["tool"].name.lower():
                results.append(tool_info)
        return results
    
    async def close_all(self):
        """关闭所有连接"""
        for client in self.clients.values():
            await client.close()

# 使用
manager = MCPManager()

# 添加多个服务器
await manager.add_server("filesystem", {
    "command": ["python", "-m", "mcp_server_filesystem"]
})
await manager.add_server("fetch", {
    "command": ["npx", "@modelcontextprotocol/server-fetch"]
})

# 搜索工具
search_tools = await manager.search_tools("read")
print(search_tools)

# 调用工具
result = await manager.call_tool(
    "filesystem", 
    "read_file", 
    {"path": "/data/config.json"}
)

8.5 敏感信息脱敏

class MCPClientWithRedaction:
    """带脱敏的MCP客户端"""
    
    def __init__(self, client, sensitive_patterns):
        self.client = client
        self.patterns = sensitive_patterns
        self.redaction_text = "[REDACTED]"
    
    def _redact(self, text: str) -> str:
        """脱敏处理"""
        import re
        for pattern in self.patterns:
            text = re.sub(pattern, self.redaction_text, text)
        return text
    
    async def call_tool(self, name: str, arguments: dict):
        """调用工具前脱敏参数"""
        # 脱敏参数
        redacted_args = {
            k: self._redact(str(v)) if isinstance(v, str) else v
            for k, v in arguments.items()
        }
        
        # 调用工具
        result = await self.client.call_tool(name, redacted_args)
        
        # 脱敏结果
        if hasattr(result, 'content'):
            for item in result.content:
                if item.type == 'text':
                    item.text = self._redact(item.text)
        
        return result

# 使用
sensitive_patterns = [
    r'\b\d{18}\b',  # 身份证
    r'\b1[3-9]\d{9}\b',  # 手机号
    r'\b[\w\.-]+@[\w\.-]+\.\w+\b'  # 邮箱
]

secure_client = MCPClientWithRedaction(client, sensitive_patterns)
result = await secure_client.call_tool(
    "read_file",
    {"path": "/data/user_info.txt"}  # 包含敏感信息
)