MCP 接入

Model Context Protocol（MCP）是一个开放协议，定义了 AI 应用与外部工具和数据源之间的标准化接口。LangChain 原生支持 MCP 集成，使 Agent 能够无缝接入 MCP 服务器提供的工具和资源。

什么是 MCP

MCP 由 Anthropic 提出并开源，旨在解决 AI 应用集成外部工具的碎片化问题。

核心概念：

• MCP 服务器：提供工具、资源和提示词的独立服务
• MCP 客户端：连接 MCP 服务器的 AI 应用端
• 工具（Tools）：可被模型调用的功能接口
• 资源（Resources）：可被读取的数据源
• 提示词（Prompts）：预定义的提示词模板

┌─────────────────┐     MCP Protocol      ┌─────────────────┐
│  LangChain Agent │ ◄──────────────────► │   MCP 服务器     │
│  (MCP 客户端)    │                       │                  │
│                  │                       │  ├── 搜索工具    │
│  create_agent()  │                       │  ├── 数据库      │
│  mcp_tools       │                       │  └── 文件系统    │
└─────────────────┘                       └─────────────────┘

设置 MCP 服务器

MCP 服务器可以是一个独立的 Python 服务或本地进程：

# mcp_server.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

server = Server("my-tools")

@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="get_current_time",
            description="获取当前时间",
            inputSchema={
                "type": "object",
                "properties": {
                    "timezone": {
                        "type": "string",
                        "description": "时区，如 Asia/Shanghai"
                    }
                }
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "get_current_time":
        from datetime import datetime
        import pytz
        tz = pytz.timezone(arguments.get("timezone", "Asia/Shanghai"))
        now = datetime.now(tz)
        return [TextContent(type="text", text=now.isoformat())]

if __name__ == "__main__":
    import anyio
    anyio.run(stdio_server(server))

在 LangChain 中连接 MCP

使用 MCP 工具

from langchain import create_agent
from langchain.chat_models import init_chat_model
from langchain.mcp import MCPTools

# 连接到 MCP 服务器并获取工具列表
mcp_tools = MCPTools.connect("python mcp_server.py")
# 或者连接远程 MCP 服务器
# mcp_tools = MCPTools.connect_http("http://localhost:8000/mcp")

# 获取 MCP 工具
tools = mcp_tools.get_tools()

# 创建 Agent 并使用 MCP 工具
agent = create_agent(
    model=init_chat_model("openai/gpt-4o"),
    tools=tools,  # 直接传入 MCP 工具
    system_prompt="你是一个助手，可以使用 MCP 提供的工具。",
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "现在上海几点了？"}]
})

MCP 资源为 RAG 数据源

from langchain.mcp import MCPResources

# 连接到 MCP 服务器并获取资源
mcp_resources = MCPResources.connect("python mcp_server.py")

# 获取资源内容
documents = mcp_resources.read_resources([
    "docs://getting-started",
    "docs://api-reference",
])

# 将 MCP 资源用于构建 RAG
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import init_embeddings

splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
chunks = splitter.split_documents(documents)

embeddings = init_embeddings("openai/text-embedding-3-small")
# 继续构建向量存储和检索器...

混合使用 MCP 工具和本地工具

from langchain import create_agent
from langchain.chat_models import init_chat_model
from langchain.mcp import MCPTools
from langchain.tools import tool

# 本地工具
@tool
def local_calculator(expression: str) -> float:
    """计算数学表达式。"""
    return eval(expression)

# MCP 工具
mcp_tools = MCPTools.connect("python mcp_server.py")
search_tool = mcp_tools.get_tool("web_search")
database_tool = mcp_tools.get_tool("query_database")

# 混合使用
agent = create_agent(
    model=init_chat_model("openai/gpt-4o"),
    tools=[local_calculator, search_tool, database_tool],
    system_prompt="你可以使用本地工具和 MCP 工具来完成任务。",
)

result = agent.invoke({
    "messages": [{"role": "user", "content": "查询数据库中的销售数据，并计算增长率"}]
})

连接多种 MCP 服务器

from langchain.mcp import MCPTools

# 连接多个 MCP 服务器
mcp_servers = [
    MCPTools.connect("python file_server.py"),     # 文件操作服务器
    MCPTools.connect("python db_server.py"),        # 数据库服务器
    MCPTools.connect("python search_server.py"),    # 搜索服务器
]

# 合并所有工具
all_tools = []
for server in mcp_servers:
    all_tools.extend(server.get_tools())

agent = create_agent(
    model=init_chat_model("openai/gpt-4o"),
    tools=all_tools,
    system_prompt="你是一个全能助手，可以访问文件和数据库。",
)

MCP 传输方式

MCP 支持多种传输协议：

from langchain.mcp import MCPTools

# STDIO 传输（本地进程）
mcp_tools = MCPTools.connect("python server.py")

# HTTP/SSE 传输（远程服务）
mcp_tools = MCPTools.connect_http("https://mcp.example.com/sse")

# WebSocket 传输
mcp_tools = MCPTools.connect_ws("wss://mcp.example.com/ws")

自定义 MCP 配置

from langchain.mcp import MCPTools

mcp_tools = MCPTools.connect(
    "python server.py",
    config={
        "timeout": 30,           # 超时时间（秒）
        "max_retries": 3,        # 最大重试次数
        "cache_ttl": 300,        # 工具列表缓存时间
        "allowed_tools": [       # 仅允许特定工具
            "web_search",
            "fetch_url",
        ],
        "environment": {         # 传递给服务器的环境变量
            "API_KEY": "xxx",
        },
    }
)

MCP 生态

MCP 社区提供了丰富的预构建服务器：

服务器	功能	地址
Filesystem	文件系统操作	内置
GitHub	GitHub API 工具	内置
PostgreSQL	数据库查询	内置
Brave Search	网络搜索	内置
Puppeteer	浏览器自动化	内置
Slack	Slack 集成	内置
Sqlite	SQLite 操作	内置

最佳实践

1. 使用 STDIO 传输用于本地工具，HTTP/SSE 用于远程服务
2. 缓存工具列表，避免每次调用都重新枚举
3. 设置超时和重试，防止 MCP 服务器故障影响 Agent
4. 限制允许的工具，仅暴露必要的功能
5. 监控 MCP 调用，记录使用情况便于调试
6. 容器化部署MCP 服务器，保持隔离和可移植性

下一步

• 了解Runtime 与上下文如何与 MCP 交互
• 学习Guardrails 安全护栏保护 MCP 工具调用
• 查看项目结构建议集成 MCP 的项目布局