模型上下文协议 (MCP)

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": "(3 + 5) x 12 等于多少？"}]}
    )
    weather_response = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "纽约的天气怎么样？"}]}
    )
    print(math_response)
    print(weather_response)

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

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": "纽约的天气怎么样？"})

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
    )

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

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"调用工具: {request.name}，参数: {request.args}")
    result = await handler(request)
    print(f"工具 {request.name} 返回: {result}")
    return result

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

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)

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("外部: 之前")
    result = await handler(request)
    print("外部: 之后")
    return result

async def inner_interceptor(request, handler):
    print("内部: 之前")
    result = await handler(request)
    print("内部: 之后")
    return result

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

# 执行顺序:
# 外部: 之前 -> 内部: 之前 -> 工具执行 -> 内部: 之后 -> 外部: 之后

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"工具 {request.name} 失败 (尝试 {attempt + 1})，将在 {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"工具 {request.name} 超时。请稍后重试。"
    except ConnectionError:
        return f"无法连接到 {request.name} 服务。使用缓存数据。"

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}] 进度: {percent:.1f}% - {message}")

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

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

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"请提供 {name} 的资料详情:",
        schema=UserDetails,
    )
    if result.action == "accept" and result.data:
        return f"已为 {name} 创建资料: email={result.data.email}, age={result.data.age}"
    if result.action == "decline":
        return f"用户拒绝。已为 {name} 创建最小资料。"
    return "资料创建已取消。"

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

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

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

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

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