模型上下文协议（MCP） | LangChain 中文文档

模型上下文协议（Model Context Protocol，MCP）是一种开放协议，用于标准化应用程序如何向 LLM 提供工具和上下文。LangChain 智能体可以使用 langchain-mcp-adapters 库来使用 MCP 服务器上定义的工具。模型上下文协议（Model Context Protocol，MCP）是一种开放协议，用于标准化应用程序如何向 LLM 提供工具和上下文。LangChain 智能体可以使用 @langchain/mcp-adapters 库来使用 MCP 服务器上定义的工具。

快速入门

安装 langchain-mcp-adapters 库：

Python · pipPython · uv

bash

pip install langchain-mcp-adapters

bash

uv add langchain-mcp-adapters

langchain-mcp-adapters 使智能体能够使用定义在一个或多个 MCP 服务器上的工具。

INFO

MultiServerMCPClient 默认是无状态的。每次工具调用都会创建一个全新的 MCP ClientSession，执行工具，然后进行清理。更多细节请参阅有状态会话部分。

python

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient  
from langchain.agents import create_agent

async def main():
    client = MultiServerMCPClient(  
        {
            "math": {
                "transport": "stdio",  # 本地子进程通信
                "command": "python",
                # math_server.py 文件的绝对路径
                "args": ["/path/to/math_server.py"],
            },
            "weather": {
                "transport": "http",  # 基于 HTTP 的远程服务器
                # 确保你在端口 8000 上启动天气服务器
                "url": "http://localhost:8000/mcp",
            }
        }
    )

    tools = await client.get_tools()  
    agent = create_agent(
        "claude-sonnet-4-6",
        tools  
    )
    math_response = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "what's (3 + 5) x 12?"}]}
    )
    weather_response = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "what is the weather in nyc?"}]}
    )
    print(math_response)
    print(weather_response)

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

安装 @langchain/mcp-adapters 库：

TypeScript · npmTypeScript · pnpmTypeScript · yarnTypeScript · bun

bash

npm install @langchain/mcp-adapters

bash

pnpm add @langchain/mcp-adapters

bash

yarn add @langchain/mcp-adapters

bash

bun add @langchain/mcp-adapters

@langchain/mcp-adapters 使智能体能够使用定义在一个或多个 MCP 服务器上的工具。

INFO

MultiServerMCPClient 默认是无状态的。每次工具调用都会创建一个全新的 MCP ClientSession，执行工具，然后进行清理。

import { MultiServerMCPClient } from "@langchain/mcp-adapters";  
import { ChatAnthropic } from "@langchain/anthropic";
import { createAgent } from "langchain";

const client = new MultiServerMCPClient({  
    math: {
        transport: "stdio",  // 本地子进程通信
        command: "node",
        // 替换为 math_server.js 文件的绝对路径
        args: ["/path/to/math_server.js"],
    },
    weather: {
        transport: "http",  // 基于 HTTP 的远程服务器
        // 确保你在端口 8000 上启动天气服务器
        url: "http://localhost:8000/mcp",
    },
});

const tools = await client.getTools();  
const agent = createAgent({
    model: "claude-sonnet-4-6",
    tools,  
});

const mathResponse = await agent.invoke({
    messages: [{ role: "user", content: "what's (3 + 5) x 12?" }],
});

const weatherResponse = await agent.invoke({
    messages: [{ role: "user", content: "what is the weather in nyc?" }],
});

TIP

使用 LangSmith 将 MCP 工具调用与智能体的推理步骤一起追踪。按照追踪快速入门进行设置。

自定义服务器

要创建自定义 MCP 服务器，请使用 FastMCP 库：

Python · pipPython · uv

bash

pip install fastmcp

bash

uv add fastmcp

要创建你自己的 MCP 服务器，可以使用 @modelcontextprotocol/sdk 库。该库提供了一种简单的方式，用于定义工具并将其作为服务器运行。

TypeScript · npmTypeScript · pnpmTypeScript · yarnTypeScript · bun

bash

npm install @modelcontextprotocol/sdk

bash

pnpm add @modelcontextprotocol/sdk

bash

yarn add @modelcontextprotocol/sdk

bash

bun add @modelcontextprotocol/sdk

要使用 MCP 工具服务器测试你的智能体，请使用以下示例：

Python · title="Math server (stdio transport)" icon="device-floppy"Python · title="Weather server (streamable HTTP transport)" icon="wifi"

python

from fastmcp import FastMCP

mcp = FastMCP("Math")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

@mcp.tool()
def multiply(a: int, b: int) -> int:
    """Multiply two numbers"""
    return a * b

if __name__ == "__main__":
    mcp.run(transport="stdio")

python

from fastmcp import FastMCP

mcp = FastMCP("Weather")

@mcp.tool()
async def get_weather(location: str) -> str:
    """Get weather for location."""
    return "It's always sunny in New York"

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

typescript

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
    CallToolRequestSchema,
    ListToolsRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";

const server = new Server(
    {
        name: "math-server",
        version: "0.1.0",
    },
    {
        capabilities: {
            tools: {},
        },
    }
);

server.setRequestHandler(ListToolsRequestSchema, async () => {
    return {
        tools: [
        {
            name: "add",
            description: "Add two numbers",
            inputSchema: {
                type: "object",
                properties: {
                    a: {
                        type: "number",
                        description: "First number",
                    },
                    b: {
                        type: "number",
                        description: "Second number",
                    },
                },
                required: ["a", "b"],
            },
        },
        {
            name: "multiply",
            description: "Multiply two numbers",
            inputSchema: {
                type: "object",
                properties: {
                    a: {
                        type: "number",
                        description: "First number",
                    },
                    b: {
                        type: "number",
                        description: "Second number",
                    },
                },
                required: ["a", "b"],
            },
        },
        ],
    };
});

server.setRequestHandler(CallToolRequestSchema, async (request) => {
    switch (request.params.name) {
        case "add": {
            const { a, b } = request.params.arguments as { a: number; b: number };
            return {
                content: [
                {
                    type: "text",
                    text: String(a + b),
                },
                ],
            };
        }
        case "multiply": {
            const { a, b } = request.params.arguments as { a: number; b: number };
            return {
                content: [
                {
                    type: "text",
                    text: String(a * b),
                },
                ],
            };
        }
        default:
            throw new Error(`Unknown tool: ${request.params.name}`);
    }
});

async function main() {
    const transport = new StdioServerTransport();
    await server.connect(transport);
    console.error("Math MCP server running on stdio");
}

main();

typescript

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js";
import {
    CallToolRequestSchema,
    ListToolsRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";
import express from "express";

const app = express();
app.use(express.json());

const server = new Server(
    {
        name: "weather-server",
        version: "0.1.0",
    },
    {
        capabilities: {
            tools: {},
        },
    }
);

server.setRequestHandler(ListToolsRequestSchema, async () => {
    return {
        tools: [
        {
            name: "get_weather",
            description: "Get weather for location",
            inputSchema: {
            type: "object",
            properties: {
                location: {
                type: "string",
                description: "Location to get weather for",
                },
            },
            required: ["location"],
            },
        },
        ],
    };
});

server.setRequestHandler(CallToolRequestSchema, async (request) => {
    switch (request.params.name) {
        case "get_weather": {
            const { location } = request.params.arguments as { location: string };
            return {
                content: [
                    {
                        type: "text",
                        text: `It's always sunny in ${location}`,
                    },
                ],
            };
        }
        default:
            throw new Error(`Unknown tool: ${request.params.name}`);
    }
});

app.post("/mcp", async (req, res) => {
    const transport = new SSEServerTransport("/mcp", res);
    await server.connect(transport);
});

const PORT = process.env.PORT || 8000;
app.listen(PORT, () => {
    console.log(`Weather MCP server running on port ${PORT}`);
});

传输方式（Transports）

MCP 支持不同的传输机制来进行客户端与服务器之间的通信。

HTTP

http 传输方式（也称为 streamable-http）使用 HTTP 请求进行客户端与服务器之间的通信。更多细节请参阅 MCP HTTP 传输规范。

对于你自己运行的服务器，请使用本地 URL；或者使用托管 URL，例如 LangChain 文档 MCP 服务器（https://docs.langchain.com/mcp），它是公开的，不需要 API 密钥。

python

from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient

client = MultiServerMCPClient(
    {
        "mcp": {
            "transport": "http",
            # "url": "http://localhost:8000/mcp",  # 本地服务器
            "url": "https://docs.langchain.com/mcp",  # 托管服务器
        }
    }
)
tools = await client.get_tools()
agent = create_agent("openai:gpt-5.4", tools)
response = await agent.ainvoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "How do I connect LangChain to an MCP server over HTTP?",
            }
        ]
    }
)

typescript

import { MultiServerMCPClient } from "@langchain/mcp-adapters";
import { createAgent } from "langchain";

const client = new MultiServerMCPClient({
    mcp: {
        transport: "http",
        // url: "http://localhost:8000/mcp", // 本地服务器
        url: "https://docs.langchain.com/mcp", // 托管服务器
    },
});

const tools = await client.getTools();
const agent = createAgent({ model: "openai:gpt-5.4", tools });
const response = await agent.invoke({
    messages: [
        {
            role: "user",
            content: "How do I connect LangChain to an MCP server over HTTP?",
        },
    ],
});

传递请求头

通过 HTTP 连接 MCP 服务器时，可以使用连接配置中的 headers 字段来包含自定义请求头（例如用于身份验证或追踪）。这在 sse（已被 MCP 规范弃用）和 streamable_http 传输方式中受支持。

python

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent

client = MultiServerMCPClient(
    {
        "weather": {
            "transport": "http",
            "url": "http://localhost:8000/mcp",
            "headers": {  
                "Authorization": "Bearer YOUR_TOKEN",  
                "X-Custom-Header": "custom-value"
            },  
        }
    }
)
tools = await client.get_tools()
agent = create_agent("openai:gpt-5.5", tools)
response = await agent.ainvoke({"messages": "what is the weather in nyc?"})

身份验证

langchain-mcp-adapters 库在底层使用官方的 MCP SDK，它允许你通过实现 httpx.Auth 接口来提供自定义的身份验证机制。

python

from langchain_mcp_adapters.client import MultiServerMCPClient

client = MultiServerMCPClient(
    {
        "weather": {
            "transport": "http",
            "url": "http://localhost:8000/mcp",
            "auth": auth, 
        }
    }
)

stdio

客户端将服务器作为子进程启动，并通过标准输入/输出来通信。最适合本地工具和简单场景。

INFO

与 HTTP 传输方式不同，stdio 连接本质上是有状态的：子进程在客户端连接的整个生命周期内持续存在。不过，当在没有显式会话管理的情况下使用 MultiServerMCPClient 时，每次工具调用仍会创建一个新会话。请参阅有状态会话以了解如何管理持久化连接。

python

client = MultiServerMCPClient(
    {
        "math": {
            "transport": "stdio",
            "command": "python",
            "args": ["/path/to/math_server.py"],
        }
    }
)

typescript

const client = new MultiServerMCPClient({
    math: {
        transport: "stdio",
        command: "node",
        args: ["/path/to/math_server.js"],
    },
});

有状态会话

默认情况下，MultiServerMCPClient 是无状态的：每次工具调用都会创建一个全新的 MCP 会话，执行工具，然后进行清理。

如果你需要控制 MCP 会话的生命周期（例如，当与一个在多次工具调用之间保持上下文的有状态服务器协作时），你可以使用 client.session() 创建一个持久化的 ClientSession。

python

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.tools import load_mcp_tools
from langchain.agents import create_agent

client = MultiServerMCPClient({...})

# 显式创建会话
async with client.session("server_name") as session:  
    # 传入会话以加载工具、资源或提示词
    tools = await load_mcp_tools(session)  
    agent = create_agent(
        "google_genai:gemini-3.6-flash",
        tools
    )

核心功能

工具

工具（Tools）允许 MCP 服务器暴露可执行函数，LLM 可以调用这些函数来执行操作——例如查询数据库、调用 API 或与外部系统交互。LangChain 会将 MCP 工具转换为 LangChain 工具，使其可以直接在任何 LangChain 智能体或工作流中使用。

加载工具

使用 client.get_tools() 从 MCP 服务器检索工具，并将其传递给智能体：

python

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent

client = MultiServerMCPClient({...})
tools = await client.get_tools()  
agent = create_agent("claude-sonnet-4-6", tools)

默认情况下，当 MCP 工具失败时，错误会以 status="error" 的工具消息形式传回给模型，而不是抛出异常。这可以让智能体读取错误并重试。若要改为抛出异常，请在 MultiServerMCPClient 或 load_mcp_tools 上设置 handle_tool_errors=False。

这仅适用于工具执行错误（CallToolResult(isError=True)）。传输、会话和内容转换失败始终会抛出异常。

INFO

将 MCP 工具错误作为失败的工具消息返回，需要 langchain-mcp-adapters>=0.3.0。早期版本会抛出 ToolException。

使用 client.getTools() 从 MCP 服务器检索工具，并将其传递给智能体：

typescript

import { MultiServerMCPClient } from "@langchain/mcp-adapters";
import { createAgent } from "langchain";

const client = new MultiServerMCPClient({...});
const tools = await client.getTools();  
const agent = createAgent({ model: "claude-sonnet-4-6", tools });

当 MCP 工具执行失败（CallToolResult 的 isError: true）时，@langchain/mcp-adapters 会抛出 ToolException。请将工具调用包装在 try/catch 中以处理这些错误。与 Python 适配器不同，TypeScript 适配器不会将错误作为失败的工具消息返回给模型。

结构化内容

MCP 工具可以在人类可读的文本响应之外返回结构化内容。当一个工具除了要向模型显示的文本之外，还需要返回机器可解析的数据（如 JSON）时，这会非常有用。

当 MCP 工具返回 structuredContent 时，适配器会将其包装成 MCPToolArtifact，并作为工具的人工产物（artifact）返回。你可以通过 ToolMessage 上的 artifact 字段来访问它。你也可以使用拦截器（interceptors）来自动处理或转换结构化内容。

从人工产物中提取结构化内容

调用智能体之后，你可以从响应中的工具消息里访问结构化内容：

python

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
from langchain.messages import ToolMessage

client = MultiServerMCPClient({...})
tools = await client.get_tools()
agent = create_agent("claude-sonnet-4-6", tools)

result = await agent.ainvoke(
    {"messages": [{"role": "user", "content": "Get data from the server"}]}
)

# 从工具消息中提取结构化内容
for message in result["messages"]:
    if isinstance(message, ToolMessage) and message.artifact:
        structured_content = message.artifact["structured_content"]

通过拦截器附加结构化内容

如果你希望结构化内容在对话历史中可见（对模型可见），可以使用拦截器自动将结构化内容附加到工具结果中：

python

import json

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.interceptors import MCPToolCallRequest
from mcp.types import TextContent

async def append_structured_content(request: MCPToolCallRequest, handler):
    """Append structured content from artifact to tool message."""
    result = await handler(request)
    if result.structuredContent:
        result.content += [
            TextContent(type="text", text=json.dumps(result.structuredContent)),
        ]
    return result

client = MultiServerMCPClient({...}, tool_interceptors=[append_structured_content])

多模态工具内容

MCP 工具可以在其响应中返回多模态内容（图像、文本等）。当 MCP 服务器返回包含多个部分（例如文本和图像）的内容时，适配器会将其转换为 LangChain 的标准内容块。你可以通过 ToolMessage 上的 content_blocks 属性访问标准化后的表示：

python

from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient

async def access_multimodal_tool_content():
    client = MultiServerMCPClient({})
    tools = await client.get_tools()
    agent = create_agent("claude-sonnet-4-6", tools)

    result = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "Take a screenshot of the current page"}]}
    )

    # 从工具消息中访问多模态内容
    for message in result["messages"]:
        if message.type == "tool":
            # 提供商原生格式的原始内容
            print(f"Raw content: {message.content}")

            # 标准化的内容块  #
            for block in message.content_blocks:  
                if block["type"] == "text":  
                    print(f"Text: {block['text']}")  
                elif block["type"] == "image":  
                    print(f"Image URL: {block.get('url')}")  
                    print(f"Image base64: {block.get('base64', '')[:50]}...")

这使你可以以与提供商无关的方式处理多模态工具响应，无论底层 MCP 服务器如何格式化其内容。

多模态工具内容

MCP 工具可以在其响应中返回多模态内容（图像、文本等）。当 MCP 服务器返回包含多个部分（例如文本和图像）的内容时，适配器会将其转换为 LangChain 的标准内容块。你可以通过 ToolMessage 上的 contentBlocks 属性访问标准化后的表示：

import { createAgent } from "langchain";

async function accessMultimodalToolContent(): Promise<void> {
  const { MultiServerMCPClient } = await import("@langchain/mcp-adapters");
  const client = new MultiServerMCPClient({});
  const tools = await client.getTools();
  const agent = createAgent({ model: "google-genai:gemini-3.6-flash", tools });

  const result = await agent.invoke({
    messages: [
      { role: "user", content: "Take a screenshot of the current page" },
    ],
  });

  // 从工具消息中访问多模态内容
  for (const message of result.messages) {
    if (message.type === "tool") {
      // 提供商原生格式的原始内容
      console.log(`Raw content: ${message.content}`);

      // 标准化的内容块
      for (const block of message.contentBlocks) {
        if (block.type === "text") {
          console.log(`Text: ${block.text}`); 
        } else if (block.type === "image") {
          console.log(`Image URL: ${block.url}`); 
          console.log(`Image base64: ${block.base64?.slice(0, 50)}...`); 
        }
      }
    }
  }
}

import { createAgent } from "langchain";

async function accessMultimodalToolContent(): Promise<void> {
  const { MultiServerMCPClient } = await import("@langchain/mcp-adapters");
  const client = new MultiServerMCPClient({});
  const tools = await client.getTools();
  const agent = createAgent({ model: "openai:gpt-5.5", tools });

  const result = await agent.invoke({
    messages: [
      { role: "user", content: "Take a screenshot of the current page" },
    ],
  });

  // 从工具消息中访问多模态内容
  for (const message of result.messages) {
    if (message.type === "tool") {
      // 提供商原生格式的原始内容
      console.log(`Raw content: ${message.content}`);

      // 标准化的内容块
      for (const block of message.contentBlocks) {
        if (block.type === "text") {
          console.log(`Text: ${block.text}`); 
        } else if (block.type === "image") {
          console.log(`Image URL: ${block.url}`); 
          console.log(`Image base64: ${block.base64?.slice(0, 50)}...`); 
        }
      }
    }
  }
}

import { createAgent } from "langchain";

async function accessMultimodalToolContent(): Promise<void> {
  const { MultiServerMCPClient } = await import("@langchain/mcp-adapters");
  const client = new MultiServerMCPClient({});
  const tools = await client.getTools();
  const agent = createAgent({ model: "anthropic:claude-sonnet-4-6", tools });

  const result = await agent.invoke({
    messages: [
      { role: "user", content: "Take a screenshot of the current page" },
    ],
  });

  // 从工具消息中访问多模态内容
  for (const message of result.messages) {
    if (message.type === "tool") {
      // 提供商原生格式的原始内容
      console.log(`Raw content: ${message.content}`);

      // 标准化的内容块
      for (const block of message.contentBlocks) {
        if (block.type === "text") {
          console.log(`Text: ${block.text}`); 
        } else if (block.type === "image") {
          console.log(`Image URL: ${block.url}`); 
          console.log(`Image base64: ${block.base64?.slice(0, 50)}...`); 
        }
      }
    }
  }
}

import { createAgent } from "langchain";

async function accessMultimodalToolContent(): Promise<void> {
  const { MultiServerMCPClient } = await import("@langchain/mcp-adapters");
  const client = new MultiServerMCPClient({});
  const tools = await client.getTools();
  const agent = createAgent({ model: "openrouter:openrouter:z-ai/glm-5.2", tools });

  const result = await agent.invoke({
    messages: [
      { role: "user", content: "Take a screenshot of the current page" },
    ],
  });

  // 从工具消息中访问多模态内容
  for (const message of result.messages) {
    if (message.type === "tool") {
      // 提供商原生格式的原始内容
      console.log(`Raw content: ${message.content}`);

      // 标准化的内容块
      for (const block of message.contentBlocks) {
        if (block.type === "text") {
          console.log(`Text: ${block.text}`); 
        } else if (block.type === "image") {
          console.log(`Image URL: ${block.url}`); 
          console.log(`Image base64: ${block.base64?.slice(0, 50)}...`); 
        }
      }
    }
  }
}

import { createAgent } from "langchain";

async function accessMultimodalToolContent(): Promise<void> {
  const { MultiServerMCPClient } = await import("@langchain/mcp-adapters");
  const client = new MultiServerMCPClient({});
  const tools = await client.getTools();
  const agent = createAgent({ model: "fireworks:accounts/fireworks/models/glm-5p2", tools });

  const result = await agent.invoke({
    messages: [
      { role: "user", content: "Take a screenshot of the current page" },
    ],
  });

  // 从工具消息中访问多模态内容
  for (const message of result.messages) {
    if (message.type === "tool") {
      // 提供商原生格式的原始内容
      console.log(`Raw content: ${message.content}`);

      // 标准化的内容块
      for (const block of message.contentBlocks) {
        if (block.type === "text") {
          console.log(`Text: ${block.text}`); 
        } else if (block.type === "image") {
          console.log(`Image URL: ${block.url}`); 
          console.log(`Image base64: ${block.base64?.slice(0, 50)}...`); 
        }
      }
    }
  }
}

import { createAgent } from "langchain";

async function accessMultimodalToolContent(): Promise<void> {
  const { MultiServerMCPClient } = await import("@langchain/mcp-adapters");
  const client = new MultiServerMCPClient({});
  const tools = await client.getTools();
  const agent = createAgent({ model: "baseten:zai-org/GLM-5.2", tools });

  const result = await agent.invoke({
    messages: [
      { role: "user", content: "Take a screenshot of the current page" },
    ],
  });

  // 从工具消息中访问多模态内容
  for (const message of result.messages) {
    if (message.type === "tool") {
      // 提供商原生格式的原始内容
      console.log(`Raw content: ${message.content}`);

      // 标准化的内容块
      for (const block of message.contentBlocks) {
        if (block.type === "text") {
          console.log(`Text: ${block.text}`); 
        } else if (block.type === "image") {
          console.log(`Image URL: ${block.url}`); 
          console.log(`Image base64: ${block.base64?.slice(0, 50)}...`); 
        }
      }
    }
  }
}

import { createAgent } from "langchain";

async function accessMultimodalToolContent(): Promise<void> {
  const { MultiServerMCPClient } = await import("@langchain/mcp-adapters");
  const client = new MultiServerMCPClient({});
  const tools = await client.getTools();
  const agent = createAgent({ model: "ollama:north-mini-code-1.0", tools });

  const result = await agent.invoke({
    messages: [
      { role: "user", content: "Take a screenshot of the current page" },
    ],
  });

  // 从工具消息中访问多模态内容
  for (const message of result.messages) {
    if (message.type === "tool") {
      // 提供商原生格式的原始内容
      console.log(`Raw content: ${message.content}`);

      // 标准化的内容块
      for (const block of message.contentBlocks) {
        if (block.type === "text") {
          console.log(`Text: ${block.text}`); 
        } else if (block.type === "image") {
          console.log(`Image URL: ${block.url}`); 
          console.log(`Image base64: ${block.base64?.slice(0, 50)}...`); 
        }
      }
    }
  }
}

这使你可以以与提供商无关的方式处理多模态工具响应，无论底层 MCP 服务器如何格式化其内容。

资源（Resources）

资源（Resources）允许 MCP 服务器暴露可供客户端读取的数据——例如文件、数据库记录或 API 响应。LangChain 会将 MCP 资源转换为 Blob 对象，这些对象提供了统一的接口来处理文本和二进制内容。

加载资源

使用 client.get_resources() 从 MCP 服务器加载资源：

python

from langchain_mcp_adapters.client import MultiServerMCPClient

client = MultiServerMCPClient({...})

# 从服务器加载所有资源
blobs = await client.get_resources("server_name")  

# 或者按 URI 加载特定资源
blobs = await client.get_resources("server_name", uris=["file:///path/to/file.txt"])  

for blob in blobs:
    print(f"URI: {blob.metadata['uri']}, MIME type: {blob.mimetype}")
    print(blob.as_string())  # 用于文本内容

你也可以将 load_mcp_resources 与会话直接配合使用，以获得更多控制：

python

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.resources import load_mcp_resources

client = MultiServerMCPClient({...})

async with client.session("server_name") as session:
    # 加载所有资源
    blobs = await load_mcp_resources(session)

    # 或者按 URI 加载特定资源
    blobs = await load_mcp_resources(session, uris=["file:///path/to/file.txt"])

提示词（Prompts）

提示词（Prompts）允许 MCP 服务器暴露可复用的提示词模板，客户端可以获取并使用这些模板。LangChain 会将 MCP 提示词转换为消息，使其易于集成到基于聊天的工作流中。

加载提示词

使用 client.get_prompt() 从 MCP 服务器加载提示词：

python

from langchain_mcp_adapters.client import MultiServerMCPClient

client = MultiServerMCPClient({...})

# 按名称加载提示词
messages = await client.get_prompt("server_name", "summarize")  

# 带参数加载提示词
messages = await client.get_prompt(  
    "server_name",  
    "code_review",  
    arguments={"language": "python", "focus": "security"}  
)  

# 在你的工作流中使用这些消息
for message in messages:
    print(f"{message.type}: {message.content}")

你也可以将 load_mcp_prompt 与会话直接配合使用，以获得更多控制：

python

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.prompts import load_mcp_prompt

client = MultiServerMCPClient({...})

async with client.session("server_name") as session:
    # 按名称加载提示词
    messages = await load_mcp_prompt(session, "summarize")

    # 带参数加载提示词
    messages = await load_mcp_prompt(
        session,
        "code_review",
        arguments={"language": "python", "focus": "security"}
    )

高级功能

工具拦截器

MCP 服务器作为独立进程运行——它们无法访问 LangGraph 的运行时信息，例如 store、context 或智能体状态。拦截器弥合了这一差距，让你在 MCP 工具执行期间访问这些运行时上下文。

拦截器还提供类似中间件的对工具调用的控制：你可以修改请求、实现重试、动态添加请求头，或完全短路执行。

章节	描述
访问运行时上下文	读取用户 ID、API 密钥、store 数据和智能体状态
状态更新与命令	使用 `Command` 更新智能体状态或控制图流程
编写拦截器	修改请求、组合拦截器和错误处理的模式

访问运行时上下文

当 MCP 工具在 LangChain 智能体中使用时（通过 create_agent），拦截器可以访问 ToolRuntime 上下文。这提供了对工具调用 ID、状态、config 和 store 的访问——从而支持访问用户数据、持久化信息和控制智能体行为的强大模式。

运行时上下文

访问在调用时传入的用户特定配置，例如用户 ID、API 密钥或权限：

python

from dataclasses import dataclass
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.interceptors import MCPToolCallRequest
from langchain.agents import create_agent

@dataclass
class Context:
    user_id: str
    api_key: str

async def inject_user_context(
    request: MCPToolCallRequest,
    handler,
):
    """Inject user credentials into MCP tool calls."""
    runtime = request.runtime
    user_id = runtime.context.user_id  
    api_key = runtime.context.api_key  

    # 向工具参数中添加用户上下文
    modified_request = request.override(
        args={**request.args, "user_id": user_id}
    )
    return await handler(modified_request)

client = MultiServerMCPClient(
    {...},
    tool_interceptors=[inject_user_context],
)
tools = await client.get_tools()
agent = create_agent("gpt-5.5", tools, context_schema=Context)

# 携带用户上下文调用
result = await agent.ainvoke(
    {"messages": [{"role": "user", "content": "Search my orders"}]},
    context={"user_id": "user_123", "api_key": "sk-..."}
)

Store

访问长期记忆以检索用户偏好，或在多次对话之间持久化数据：

python

from dataclasses import dataclass
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.interceptors import MCPToolCallRequest
from langchain.agents import create_agent
from langgraph.store.memory import InMemoryStore

@dataclass
class Context:
    user_id: str

async def personalize_search(
    request: MCPToolCallRequest,
    handler,
):
    """Personalize MCP tool calls using stored preferences."""
    runtime = request.runtime
    user_id = runtime.context.user_id
    store = runtime.store  

    # 从 store 读取用户偏好
    prefs = store.get(("preferences",), user_id)  

    if prefs and request.name == "search":
        # 应用用户偏好的语言和结果数量限制
        modified_args = {
            **request.args,
            "language": prefs.value.get("language", "en"),
            "limit": prefs.value.get("result_limit", 10),
        }
        request = request.override(args=modified_args)

    return await handler(request)

client = MultiServerMCPClient(
    {...},
    tool_interceptors=[personalize_search],
)
tools = await client.get_tools()
agent = create_agent(
    "gpt-5.5",
    tools,
    context_schema=Context,
    store=InMemoryStore()
)

状态

访问会话状态，以便根据当前会话做出决策：

python

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.interceptors import MCPToolCallRequest
from langchain.messages import ToolMessage

async def require_authentication(
    request: MCPToolCallRequest,
    handler,
):
    """Block sensitive MCP tools if user is not authenticated."""
    runtime = request.runtime
    state = runtime.state  
    is_authenticated = state.get("authenticated", False)  

    sensitive_tools = ["delete_file", "update_settings", "export_data"]

    if request.name in sensitive_tools and not is_authenticated:
        # 返回错误而不是调用工具
        return ToolMessage(
            content="Authentication required. Please log in first.",
            tool_call_id=runtime.tool_call_id,
        )

    return await handler(request)

client = MultiServerMCPClient(
    {...},
    tool_interceptors=[require_authentication],
)

工具调用 ID

访问工具调用 ID，以返回格式正确的响应或追踪工具执行情况：

python

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.interceptors import MCPToolCallRequest
from langchain.messages import ToolMessage

async def rate_limit_interceptor(
    request: MCPToolCallRequest,
    handler,
):
    """Rate limit expensive MCP tool calls."""
    runtime = request.runtime
    tool_call_id = runtime.tool_call_id  

    # 检查速率限制（简化示例）
    if is_rate_limited(request.name):
        return ToolMessage(
            content="Rate limit exceeded. Please try again later.",
            tool_call_id=tool_call_id,  
        )

    result = await handler(request)

    # 记录成功的工具调用
    log_tool_execution(tool_call_id, request.name, success=True)

    return result

client = MultiServerMCPClient(
    {...},
    tool_interceptors=[rate_limit_interceptor],
)

更多上下文工程模式，请参阅上下文工程和工具。

状态更新与命令

拦截器可以返回 Command 对象来更新智能体状态或控制图的执行流程。这对于跟踪任务进度、在智能体之间切换或提前结束执行非常有用。

python

from langchain.agents import AgentState, create_agent
from langchain_mcp_adapters.interceptors import MCPToolCallRequest
from langchain.messages import ToolMessage
from langgraph.types import Command

async def handle_task_completion(
    request: MCPToolCallRequest,
    handler,
):
    """Mark task complete and hand off to summary agent."""
    result = await handler(request)

    if request.name == "submit_order":
        return Command(
            update={
                "messages": [result] if isinstance(result, ToolMessage) else [],
                "task_status": "completed",  
            },
            goto="summary_agent",  
        )

    return result

使用 Command 并配合 goto="__end__" 来提前结束执行：

python

async def end_on_success(
    request: MCPToolCallRequest,
    handler,
):
    """End agent run when task is marked complete."""
    result = await handler(request)

    if request.name == "mark_complete":
        return Command(
            update={"messages": [result], "status": "done"},
            goto="__end__",  
        )

    return result

自定义拦截器

拦截器是包装工具执行的异步函数，支持请求/响应修改、重试逻辑以及其他横切关注点。它们遵循一种"洋葱"模式，列表中第一个拦截器是最外层。

基本模式

拦截器是一个接收请求和处理器的异步函数。你可以在调用处理器之前修改请求，在之后修改响应，或者完全跳过处理器。

python

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.interceptors import MCPToolCallRequest

async def logging_interceptor(
    request: MCPToolCallRequest,
    handler,
):
    """Log tool calls before and after execution."""
    print(f"Calling tool: {request.name} with args: {request.args}")
    result = await handler(request)
    print(f"Tool {request.name} returned: {result}")
    return result

client = MultiServerMCPClient(
    {"math": {"transport": "stdio", "command": "python", "args": ["/path/to/server.py"]}},
    tool_interceptors=[logging_interceptor],  
)

修改请求

使用 request.override() 来创建一个修改后的请求。这遵循不可变模式，保持原始请求不变。

python

async def double_args_interceptor(
    request: MCPToolCallRequest,
    handler,
):
    """Double all numeric arguments before execution."""
    modified_args = {k: v * 2 for k, v in request.args.items()}
    modified_request = request.override(args=modified_args)  
    return await handler(modified_request)

# 原始调用：add(a=2, b=3) 变为 add(a=4, b=6)

在运行时修改请求头

拦截器可以根据请求上下文动态修改 HTTP 请求头：

python

async def auth_header_interceptor(
    request: MCPToolCallRequest,
    handler,
):
    """Add authentication headers based on the tool being called."""
    token = get_token_for_tool(request.name)
    modified_request = request.override(
        headers={"Authorization": f"Bearer {token}"}  
    )
    return await handler(modified_request)

组合拦截器

多个拦截器按"洋葱"顺序组合——列表中的第一个拦截器是最外层：

python

async def outer_interceptor(request, handler):
    print("outer: before")
    result = await handler(request)
    print("outer: after")
    return result

async def inner_interceptor(request, handler):
    print("inner: before")
    result = await handler(request)
    print("inner: after")
    return result

client = MultiServerMCPClient(
    {...},
    tool_interceptors=[outer_interceptor, inner_interceptor],  
)

# 执行顺序：
# outer: before -> inner: before -> 工具执行 -> inner: after -> outer: after

错误处理

使用拦截器来捕获工具执行过程中抛出的异常（例如传输或运行时故障），并添加重试逻辑。工具执行错误（CallToolResult(isError=True)）默认不会抛出，因此捕获异常的拦截器永远不会触发它们。要在这里将这些错误作为异常捕获，请设置 handle_tool_errors=False。

python

import asyncio

async def retry_interceptor(
    request: MCPToolCallRequest,
    handler,
    max_retries: int = 3,
    delay: float = 1.0,
):
    """Retry failed tool calls with exponential backoff."""
    last_error = None
    for attempt in range(max_retries):
        try:
            return await handler(request)
        except Exception as e:
            last_error = e
            if attempt < max_retries - 1:
                wait_time = delay * (2 ** attempt)  # 指数退避
                print(f"Tool {request.name} failed (attempt {attempt + 1}), retrying in {wait_time}s...")
                await asyncio.sleep(wait_time)
    raise last_error

client = MultiServerMCPClient(
    {...},
    tool_interceptors=[retry_interceptor],  
)

你也可以捕获特定类型的错误并返回回退值：

python

async def fallback_interceptor(
    request: MCPToolCallRequest,
    handler,
):
    """Return a fallback value if tool execution fails."""
    try:
        return await handler(request)
    except TimeoutError:
        return f"Tool {request.name} timed out. Please try again later."
    except ConnectionError:
        return f"Could not connect to {request.name} service. Using cached data."

进度通知

订阅长时间运行的工具执行的进度更新：

python

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.callbacks import Callbacks, CallbackContext

async def on_progress(
    progress: float,
    total: float | None,
    message: str | None,
    context: CallbackContext,
):
    """Handle progress updates from MCP servers."""
    percent = (progress / total * 100) if total else progress
    tool_info = f" ({context.tool_name})" if context.tool_name else ""
    print(f"[{context.server_name}{tool_info}] Progress: {percent:.1f}% - {message}")

client = MultiServerMCPClient(
    {...},
    callbacks=Callbacks(on_progress=on_progress),  
)

CallbackContext 提供：

server_name：MCP 服务器的名称
tool_name：正在执行的工具的名称（在工具调用期间可用）

日志记录

MCP 协议支持来自服务器的日志记录通知。使用 Callbacks 类来订阅这些事件。

python

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.callbacks import Callbacks, CallbackContext
from mcp.types import LoggingMessageNotificationParams

async def on_logging_message(
    params: LoggingMessageNotificationParams,
    context: CallbackContext,
):
    """Handle log messages from MCP servers."""
    print(f"[{context.server_name}] {params.level}: {params.data}")

client = MultiServerMCPClient(
    {...},
    callbacks=Callbacks(on_logging_message=on_logging_message),  
)

征询（Elicitation）

征询（Elicitation）允许 MCP 服务器在工具执行期间向用户请求额外输入。服务器不必在一开始就要求提供所有输入，而是可以根据需要交互式地询问信息。

服务器设置

定义一个使用 ctx.elicit() 并配合 schema 来请求用户输入的工具：

python

from pydantic import BaseModel
from mcp.server.fastmcp import Context, FastMCP

server = FastMCP("Profile")

class UserDetails(BaseModel):
    email: str
    age: int

@server.tool()
async def create_profile(name: str, ctx: Context) -> str:
    """Create a user profile, requesting details via elicitation."""
    result = await ctx.elicit(  
        message=f"Please provide details for {name}'s profile:",  
        schema=UserDetails,  
    )  
    if result.action == "accept" and result.data:
        return f"Created profile for {name}: email={result.data.email}, age={result.data.age}"
    if result.action == "decline":
        return f"User declined. Created minimal profile for {name}."
    return "Profile creation cancelled."

if __name__ == "__main__":
    server.run(transport="http")

客户端设置

通过向 MultiServerMCPClient 提供一个回调来处理征询请求：

python

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.callbacks import Callbacks, CallbackContext
from mcp.shared.context import RequestContext
from mcp.types import ElicitRequestParams, ElicitResult

async def on_elicitation(
    mcp_context: RequestContext,
    params: ElicitRequestParams,
    context: CallbackContext,
) -> ElicitResult:
    """Handle elicitation requests from MCP servers."""
    # 在真实应用中，你会提示用户输入
    # 基于 params.message 和 params.requestedSchema
    return ElicitResult(  
        action="accept",  
        content={"email": "user@example.com", "age": 25},  
    )  

client = MultiServerMCPClient(
    {
        "profile": {
            "url": "http://localhost:8000/mcp",
            "transport": "http",
        }
    },
    callbacks=Callbacks(on_elicitation=on_elicitation),  
)

响应动作

征询回调可以返回三种动作之一：

动作	描述
`accept`	用户提供了有效输入。在 `content` 字段中包含该数据。
`decline`	用户选择不提供所请求的信息。
`cancel`	用户完全取消了该操作。

python

# 接受并附带数据
ElicitResult(action="accept", content={"email": "user@example.com", "age": 25})

# 拒绝（用户不想提供信息）
ElicitResult(action="decline")

# 取消（中止该操作）
ElicitResult(action="cancel")

Patterns

Integrations

Multi-agent

Test

快速入门

自定义服务器

传输方式（Transports）

HTTP

传递请求头

身份验证

stdio

有状态会话

核心功能

工具

加载工具

结构化内容

多模态工具内容

多模态工具内容

资源（Resources）

加载资源

提示词（Prompts）

加载提示词

高级功能

工具拦截器

访问运行时上下文

运行时上下文

Store

状态

工具调用 ID

状态更新与命令

自定义拦截器

进度通知

日志记录

征询（Elicitation）

服务器设置

客户端设置

响应动作

其他资源

快速入门 ​

自定义服务器 ​

传输方式（Transports） ​

HTTP ​

传递请求头 ​

身份验证 ​

stdio ​

有状态会话 ​

核心功能 ​

工具 ​

加载工具 ​

结构化内容 ​

多模态工具内容 ​

多模态工具内容 ​

资源（Resources） ​

加载资源 ​

提示词（Prompts） ​

加载提示词 ​

高级功能 ​

工具拦截器 ​

访问运行时上下文 ​

运行时上下文 ​

Store ​

状态 ​

工具调用 ID ​

状态更新与命令 ​

自定义拦截器 ​

进度通知 ​

日志记录 ​

征询（Elicitation） ​

服务器设置 ​

客户端设置 ​

响应动作 ​

其他资源 ​

快速入门

自定义服务器

传输方式（Transports）

HTTP

传递请求头

身份验证

stdio

有状态会话

核心功能

工具

加载工具

结构化内容

多模态工具内容

多模态工具内容

资源（Resources）

加载资源

提示词（Prompts）

加载提示词

高级功能

工具拦截器

访问运行时上下文

运行时上下文

Store

状态

工具调用 ID

状态更新与命令

自定义拦截器

进度通知

日志记录

征询（Elicitation）

服务器设置

客户端设置

响应动作

其他资源