一.自定义服务器

将mcp视为 AI 的“USB 接口”：无论工具是用什么语言写的，只要符合 MCP 标准，Agent 就能直接通过一个标准的“适配器”来调用它们。

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

async def main():
    client = MultiServerMCPClient(
        {
            "math": {
                "transport": "stdio",  # Local subprocess communication
                "command": "python",
                # Absolute path to your math_server.py file
                "args": ["/path/to/math_server.py"],
            },
            "weather": {
                "transport": "http",  # HTTP-based remote server
                # Ensure you start your weather server on port 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())

Math Server:

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")

weather server:

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")

当 Agent 接收到用户的指令时，它会经历以下逻辑：

• 理解意图：分析用户需要计算数学题还是查询天气。
• 选择工具：如果是数学题，Agent 会识别出 math 服务器提供的工具。
• 协议转换：langchain-mcp-adapters 会自动把 Agent 的意图包装成 MCP 协议格式发送给对应的服务器。
• 返回结果：服务器执行完毕后，结果通过 MCP 协议传回，Agent 最终汇总成自然语言回复。

---

二.传输机制 (Transports)

MCP 协议定义了标准化的通信层，使得客户端（Client）与服务器（Server）能够跨越不同的物理边界进行对话。不同的传输机制决定了数据的流动方式及其适用场景。

1. Http
   HTTP 传输（在规范中常被称为 streamable-http）主要用于客户端与远程服务器之间的通信。它允许 Agent 通过互联网或内网调用托管在云端的工具集。

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

在生产环境中，连接远程 MCP 服务器通常需要通过身份验证（Authentication）和请求追踪（Tracing）。langchain-mcp-adapters 提供了灵活的配置项，允许开发者在 HTTP 传输层注入安全凭证。

• 传递自定义 Headers
  当你需要传递简单的 API Key、Token 或用于链路追踪的 Trace-ID 时，可以在连接配置中使用 headers 字段。该功能支持 streamable_http 传输协议。

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-4.1", tools)
response = await agent.ainvoke({"messages": "what is the weather in nyc?"})

• 身份验证 (Authentication)
  在生产级应用中，工具服务器通常受到严格的访问控制。langchain-mcp-adapters 库底层集成了官方的 MCP SDK，这意味着它支持极其灵活的自定义认证方案——通过实现 httpx.Auth 接口，你可以将任何复杂的安全逻辑注入到 Agent 的工具调用过程中。如果你需要处理比静态 Header 更复杂的场景（如动态生成的签名、需要刷新权标的 OAuth2 或基于时间的哈希校验），可以通过自定义 httpx.Auth 类来实现：

from langchain_mcp_adapters.client import MultiServerMCPClient

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

2. stdio 传输 (Local Subprocess)
   stdio 传输机制是 MCP 协议中最简单且最高效的本地通信方式。在这种模式下，客户端（如 LangChain Agent）直接启动一个子进程作为服务器，并通过标准输入（stdin）和标准输出 (stdout) 与其交换 JSON-RPC 消息。

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

架构提示：stdio 是 MCP 协议中最基础且最高效的本地通信模式。与基于 Web 的 HTTP 传输不同，stdio 连接在物理层面是持久的。在这种机制下，客户端将服务器作为一个子进程启动，并利用标准输入（stdin）和标准输出（stdout）进行双向数据交换。只要客户端与服务器的连接保持开启，该子进程就会持续驻留在内存中。这避免了每次调用工具时重新加载代码、环境变量或大型 AI 模型的巨大开销。逻辑会话隔离：虽然进程在运行，但在使用 MultiServerMCPClient 且未配置显式会话管理时，每次工具调用在逻辑上仍会被视为一个“新会话”。这意味着工具在处理不同请求时，其内部状态默认是不共享的)。

---

三.有状态会话 (Stateful Sessions)

在默认情况下，MultiServerMCPClient 遵循无状态（Stateless）原则：每次工具调用都会创建一个全新的 MCP 会话，执行任务后立即清理。这种设计保证了调用的原子性和安全性，但在需要跨步操作（如：先登录、后查询）的场景下，则需要引入持久化会话。

特性	默认模式 (Stateless)	持久模式 (Stateful)
生命周期	随单次工具调用开始与结束。	由开发者手动开启 (session()) 与关闭。
上下文维护	每次调用都是“初次见面”，不保留变量。	服务器端可保留临时文件、登录状态或内存变量。
资源消耗	频繁创建/销毁连接，开销略高。	保持连接驻留，响应速度更快，但需注意资源回收。

你可以通过上下文管理器（async with）来确保会话在完成后能够正确关闭，避免资源泄露：

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

client = MultiServerMCPClient({...})

# Create a session explicitly
async with client.session("server_name") as session:
    # Pass the session to load tools, resources, or prompts
    tools = await load_mcp_tools(session)
    agent = create_agent(
        "anthropic:claude-3-7-sonnet-latest",
        tools
    )

---

四.核心功能

1. 工具 (Tools)

工具是 MCP 协议中最具行动力的部分。它允许 MCP 服务器向外界暴露可执行的函数，使 LLM 能够突破“纯文字生成”的限制，直接执行动作——如查询数据库、调用 API 或操作本地系统。

• 加载工具 (Loading Tools)
  在 MCP 架构中，工具的发现与注入是一个自动化的动态过程。通过调用 client.get_tools()，客户端会与所有配置好的服务器通信，获取其支持的函数签名和参数 Schema。

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)

• 结构化内容 (Structured Content)
  在复杂的 Agent 交互中，工具往往不仅需要返回给模型一段“人看得懂”的文本描述，还需要返回“机器可读”的原始数据（如 JSON、图片数据或二进制流）。MCP 协议通过 Structured Content 机制实现了这两者的优雅分离。

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"}]}
)

# Extract structured content from tool messages
for message in result["messages"]:
    if isinstance(message, ToolMessage) and message.artifact:
        structured_content = message.artifact["structured_content"]

• 通过拦截器追加结构化内容 (Interceptors)
  在默认情况下，结构化内容（Structured Content）存储在 artifact 字段中，模型是无法直接“读到”这些数据的。如果你希望模型能够基于这些机器数据进行推理或总结，可以使用**拦截器（Interceptor）**机制，自动将结构化内容追加到工具的文本结果中。拦截器就像是一个“中间件”，它在工具执行完毕后、结果返回给模型之前进行干预。它可以读取 artifact 中的数据，并将其转换为字符串拼接到模型的对话历史中。

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])

• 多模态工具内容 (Multimodal Tool Content)
  现代 AI 工具往往不再局限于文本输出。MCP 协议原生支持多模态响应，允许工具在一次返回中同时包含图像、文本、音频或其他媒体类型。你可以遍历 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"}]}
)

# Access multimodal content from tool messages
for message in result["messages"]:
    if message.type == "tool":
        # Raw content in provider-native format
        print(f"Raw content: {message.content}")

        # Standardized content blocks  #
        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]}...")

2. 资源 (Resources)

资源是 MCP 协议中用于“共享数据”的核心机制。如果说“工具”代表了 Agent 的手脚（执行动作），那么“资源”则代表了 Agent 的耳目。它允许 MCP 服务器暴露各种形式的数据——如本地文件、数据库记录或实时的 API 响应——供客户端（Agent）读取。
为了简化对不同数据类型的处理，langchain-mcp-adapters 会将获取到的 MCP 资源自动转换为 Blob 对象。这种设计为处理文本（Text）和二进制（Binary）内容提供了一个完全统一的界面。

特性	说明	业务价值
URI 寻址	每个资源都有唯一的 URI（如 file:///logs/app.log）。	方便 Agent 精确引用特定的数据源。
多格式支持	无论是纯文本配置还是图片/PDF 二进制流。	开发者无需为不同文件类型编写多套解析代码。
按需读取	Agent 可以根据需要决定是否加载资源内容。	优化上下文利用率，避免一次性注入过多无关数据。

from langchain_mcp_adapters.client import MultiServerMCPClient

client = MultiServerMCPClient({...})

# Load all resources from a server
blobs = await client.get_resources("server_name")

# Or load specific resources by 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())  # For text content

也可以使用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:
    # Load all resources
    blobs = await load_mcp_resources(session)

    # Or load specific resources by URI
    blobs = await load_mcp_resources(session, uris=["file:///path/to/file.txt"])

3. 提示词模板 (Prompts)

提示词 (Prompts) 是 MCP 协议中用于标准化“引导逻辑”的核心机制。它允许服务器预设一套可复用的模板，并由客户端（Agent）根据当前任务动态填充参数。

from langchain_mcp_adapters.client import MultiServerMCPClient

client = MultiServerMCPClient({...})

# 从指定的服务器（server_name）中请求名为 summarize 的预设提示词
messages = await client.get_prompt("server_name", "summarize")

# Load a prompt with arguments
messages = await client.get_prompt(
    "server_name",
    "code_review",
    arguments={"language": "python", "focus": "security"}
)

# Use the messages in your workflow
for message in messages:
    print(f"{message.type}: {message.content}")

• 从指定的服务器（server_name）中请求名为 summarize 的预设提示词。
• 服务器端的 code_review 可能定义为："请作为一名 {{language}} 专家，重点针对 {{focus}} 方面进行代码审查。
• 通过 arguments 字典，适配器会自动完成占位符的填充，确保返回的消息内容已经过参数化处理。

五. 工具拦截器 (Tool Interceptors)

工具拦截器是 MCP 架构中的“黑盒子解码器”。由于 MCP 服务器运行在独立的进程中，它们天然无法访问 LangGraph 的运行时信息（如 store、context 或 state）。拦截器的存在正是为了桥接这一鸿沟，将动态的运行时上下文注入到静态的工具执行流程中。拦截器在工具请求发出之前和结果返回之后发挥作用。你可以将其视为一个功能强大的中间件层。

职责	技术细节	典型场景
运行时上下文注入	将 LangGraph 的 state 或 user_id 注入到工具参数中。	工具需要根据当前用户的历史偏好调整执行逻辑。
请求/响应重塑	动态修改请求 Header、参数，或对返回结果进行清洗脱敏。	在数据返回给模型前，自动屏蔽掉敏感的个人身份信息。
执行流控制	实现自动重试逻辑、动态鉴权或直接“熔断”非法请求。	当检测到工具调用频率过高时，直接返回自定义错误而不触发进程。

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  

    # Add user context to tool arguments
    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-4.1", tools, context_schema=Context)

# Invoke with user context
result = await agent.ainvoke(
    {"messages": [{"role": "user", "content": "Search my orders"}]},
    context={"user_id": "user_123", "api_key": "sk-..."}
)

MCP 是 AI 时代的“USB 标准接口”，它通过统一的协议将 LLM 与外部世界连接起来。 其核心架构由工具 (Tools)、资源 (Resources) 和提示词 (Prompts) 三位一体构成：工具赋予 Agent 执行动作的能力（如查库、调 API），资源提供静态数据访问（如日志、文件），而提示词模板则实现了引导逻辑的工程化管理。通过 stdio 或 HTTP 传输机制，MCP 支持从简单的本地子进程到复杂的云端部署；配合拦截器 (Interceptors) 和持久化会话，它能无缝桥接 AI 运行时上下文，实现高效、解耦且具备多模态能力的自动化工作流。