Documentation Index
Fetch the complete documentation index at: https://nvd-54.mintlify.app/llms.txt
Use this file to discover all available pages before exploring further.
模型上下文协议(MCP)是一个开放协议,标准化了应用程序向大语言模型(LLM)提供工具和上下文的方式。LangChain 智能体可以通过 langchain-mcp-adapters 库使用 MCP 服务器上定义的工具。
快速入门
安装 langchain-mcp-adapters 库:
pip install langchain-mcp-adapters
langchain-mcp-adapters 允许智能体使用一个或多个 MCP 服务器上定义的工具。
MultiServerMCPClient 默认是无状态的。每次工具调用都会创建一个新的 MCP ClientSession,执行工具,然后清理。有关更多详情,请参阅有状态会话部分。
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())
自定义服务器
要创建自定义 MCP 服务器,请使用 FastMCP 库:
要使用 MCP 工具服务器测试你的智能体,请使用以下示例:
from fastmcp import FastMCP
mcp = FastMCP("Math")
@mcp.tool()
def add(a: int, b: int) -> int:
"""两数相加"""
return a + b
@mcp.tool()
def multiply(a: int, b: int) -> int:
"""两数相乘"""
return a * b
if __name__ == "__main__":
mcp.run(transport="stdio")
传输方式
MCP 支持不同的客户端-服务器通信传输机制。
HTTP
http 传输(也称为 streamable-http)使用 HTTP 请求进行客户端-服务器通信。有关更多详情,请参阅 MCP HTTP 传输规范。
client = MultiServerMCPClient(
{
"weather": {
"transport": "http",
"url": "http://localhost:8000/mcp",
}
}
)
传递请求头
通过 HTTP 连接到 MCP 服务器时,你可以使用连接配置中的 headers 字段包含自定义请求头(例如用于认证或追踪)。sse(已被 MCP 规范弃用)和 streamable_http 传输均支持此功能。
使用 MultiServerMCPClient 传递请求头
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.4", tools)
response = await agent.ainvoke({"messages": "what is the weather in nyc?"})
langchain-mcp-adapters 库底层使用官方 MCP SDK,允许你通过实现 httpx.Auth 接口来提供自定义认证机制。
from langchain_mcp_adapters.client import MultiServerMCPClient
client = MultiServerMCPClient(
{
"weather": {
"transport": "http",
"url": "http://localhost:8000/mcp",
"auth": auth,
}
}
)
stdio
客户端将服务器作为子进程启动,并通过标准输入/输出进行通信。最适合本地工具和简单配置。
与 HTTP 传输不同,stdio 连接本质上是有状态的:子进程在客户端连接的整个生命周期内持续存在。但是,当使用 MultiServerMCPClient 而不进行显式会话管理时,每次工具调用仍然会创建一个新会话。有关管理持久连接的信息,请参阅有状态会话。
client = MultiServerMCPClient(
{
"math": {
"transport": "stdio",
"command": "python",
"args": ["/path/to/math_server.py"],
}
}
)
有状态会话
默认情况下,MultiServerMCPClient 是无状态的:每次工具调用都会创建一个新的 MCP 会话,执行工具,然后清理。
如果你需要控制 MCP 会话的生命周期(例如,当使用跨工具调用维护上下文的有状态服务器时),你可以使用 client.session() 创建持久的 ClientSession。
使用 MCP ClientSession 进行有状态工具使用
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.1-pro-preview",
tools
)
核心功能
工具允许 MCP 服务器暴露可执行函数,供大语言模型(LLM)调用以执行操作——如查询数据库、调用 API 或与外部系统交互。LangChain 将 MCP 工具转换为 LangChain 工具,使其可以直接在任何 LangChain 智能体或工作流中使用。
加载工具
使用 client.get_tools() 从 MCP 服务器检索工具并传递给你的智能体:
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 工具可以在人类可读的文本响应之外返回结构化内容。当工具需要返回机器可解析的数据(如 JSON)以及展示给模型的文本时,这很有用。
当 MCP 工具返回 structuredContent 时,适配器将其包装在 MCPToolArtifact 中,并作为工具的 artifact 返回。你可以通过 ToolMessage 上的 artifact 字段访问它。你也可以使用拦截器自动处理或转换结构化内容。
从 artifact 提取结构化内容
调用智能体后,你可以从响应中的工具消息访问结构化内容:
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"]
通过拦截器追加结构化内容
如果你希望结构化内容在对话历史中可见(对模型可见),你可以使用拦截器自动将结构化内容追加到工具结果中:
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):
"""将 artifact 中的结构化内容追加到工具消息。"""
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 属性访问标准化表示:
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)
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 服务器暴露数据——如文件、数据库记录或 API 响应——供客户端读取。LangChain 将 MCP 资源转换为 Blob 对象,提供统一的文本和二进制内容处理接口。
加载资源
使用 client.get_resources() 从 MCP 服务器加载资源:
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 配合会话进行更精细的控制:
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"])
提示词
提示词允许 MCP 服务器暴露可复用的提示词模板,供客户端检索和使用。LangChain 将 MCP 提示词转换为消息,使其易于集成到基于聊天的工作流中。
加载提示词
使用 client.get_prompt() 从 MCP 服务器加载提示词:
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 配合会话进行更精细的控制:
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 运行时信息,如存储、上下文或智能体状态。拦截器弥补了这个差距,让你在 MCP 工具执行期间可以访问这些运行时上下文。
拦截器还提供类似中间件的工具调用控制:你可以修改请求、实现重试、动态添加请求头,或完全短路执行。
| 部分 | 描述 |
|---|
| 访问运行时上下文 | 读取用户 ID、API 密钥、存储数据和智能体状态 |
| 状态更新和命令 | 更新智能体状态或使用 Command 控制图流程 |
| 编写拦截器 | 修改请求、组合拦截器和错误处理的模式 |
访问运行时上下文
当 MCP 工具在 LangChain 智能体中使用时(通过 create_agent),拦截器可以访问 ToolRuntime 上下文。这提供了对工具调用 ID、状态、配置和存储的访问——支持访问用户数据、持久化信息和控制智能体行为等强大模式。
访问用户特定的配置,如在调用时传递的用户 ID、API 密钥或权限: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,
):
"""将用户凭据注入 MCP 工具调用。"""
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.4", tools, context_schema=Context)
# 使用用户上下文调用
result = await agent.ainvoke(
{"messages": [{"role": "user", "content": "Search my orders"}]},
context={"user_id": "user_123", "api_key": "sk-..."}
)
访问长期记忆以检索用户偏好或跨对话持久化数据: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,
):
"""使用存储的偏好个性化 MCP 工具调用。"""
runtime = request.runtime
user_id = runtime.context.user_id
store = runtime.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.4",
tools,
context_schema=Context,
store=InMemoryStore()
)
访问对话状态以根据当前会话做出决策: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,
):
"""如果用户未认证,则阻止敏感 MCP 工具。"""
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 以返回格式正确的响应或追踪工具执行: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,
):
"""对昂贵的 MCP 工具调用进行速率限制。"""
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 对象来更新智能体状态或控制图执行流程。这对于追踪任务进度、在智能体之间切换或提前结束执行很有用。
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,
):
"""标记任务完成并移交给摘要智能体。"""
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__" 提前结束执行:
async def end_on_success(
request: MCPToolCallRequest,
handler,
):
"""当任务标记为完成时结束智能体运行。"""
result = await handler(request)
if request.name == "mark_complete":
return Command(
update={"messages": [result], "status": "done"},
goto="__end__",
)
return result
自定义拦截器
拦截器是包装工具执行的异步函数,支持请求/响应修改、重试逻辑和其他横切关注点。它们遵循”洋葱”模式,列表中的第一个拦截器是最外层。
基本模式
拦截器是一个接收请求和处理器的异步函数。你可以在调用处理器之前修改请求、在之后修改响应,或完全跳过处理器。
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.interceptors import MCPToolCallRequest
async def logging_interceptor(
request: MCPToolCallRequest,
handler,
):
"""在执行前后记录工具调用日志。"""
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() 创建修改后的请求。这遵循不可变模式,原始请求保持不变。
async def double_args_interceptor(
request: MCPToolCallRequest,
handler,
):
"""在执行前将所有数值参数翻倍。"""
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 请求头:
async def auth_header_interceptor(
request: MCPToolCallRequest,
handler,
):
"""根据被调用的工具添加认证请求头。"""
token = get_token_for_tool(request.name)
modified_request = request.override(
headers={"Authorization": f"Bearer {token}"}
)
return await handler(modified_request)
组合拦截器
多个拦截器以”洋葱”顺序组合——列表中的第一个拦截器是最外层:
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
错误处理
使用拦截器捕获工具执行错误并实现重试逻辑:
import asyncio
async def retry_interceptor(
request: MCPToolCallRequest,
handler,
max_retries: int = 3,
delay: float = 1.0,
):
"""使用指数退避重试失败的工具调用。"""
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],
)
你也可以捕获特定的错误类型并返回降级值:
async def fallback_interceptor(
request: MCPToolCallRequest,
handler,
):
"""如果工具执行失败则返回降级值。"""
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."
进度通知
订阅长时间运行的工具执行的进度更新:
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,
):
"""处理来自 MCP 服务器的进度更新。"""
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 类订阅这些事件。
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,
):
"""处理来自 MCP 服务器的日志消息。"""
print(f"[{context.server_name}] {params.level}: {params.data}")
client = MultiServerMCPClient(
{...},
callbacks=Callbacks(on_logging_message=on_logging_message),
)
引出允许 MCP 服务器在工具执行期间请求用户的额外输入。服务器可以在需要时交互式地请求信息,而不是要求预先提供所有输入。
服务器设置
定义一个使用 ctx.elicit() 请求用户输入的工具,带有 schema:
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:
"""创建用户配置文件,通过引出请求详细信息。"""
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 提供回调来处理引出请求:
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:
"""处理来自 MCP 服务器的引出请求。"""
# 在实际应用中,你会根据 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 | 用户完全取消了操作。 |
# 接受并提供数据
ElicitResult(action="accept", content={"email": "user@example.com", "age": 25})
# 拒绝(用户不想提供信息)
ElicitResult(action="decline")
# 取消(中止操作)
ElicitResult(action="cancel")
更多资源
将这些文档连接到 Claude、VSCode 等工具,通过 MCP 获取实时答案。