LangChain-agent:agent基本能力及MCP快速上手

一.基本能力

1.1动态指定模型/工具或系统提示词

这些动态指定的方式其实很简单，但是如果按照我们的一贯思维，因为agent是基于LangGraph的，就像修改state那样，直接修改request就可以达成动态指定的效果，那就错了，我们需要借助request提供的override进行修改，比如我们看一个动态指定模型的例子:

1.1.1动态指定模型例子

from typing import Callable
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from langchain_deepseek import ChatDeepSeek

mini_model = ChatDeepSeek(model="deepseek-v4-flash",base_url="https://api.siliconflow.cn/v1")
model = ChatDeepSeek(model="deepseek-v4-pro",base_url="https://api.siliconflow.cn/v1")

@tool
def get_location() :
    '''获取用户位置的函数'''
    return "北京"

@wrap_model_call
def change_model(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse]
)-> ModelResponse | None:
    message_sumlen = len(request.messages)
    if message_sumlen >= 2:
        final_model = model
    else:
        final_model = mini_model
    return handler(request.override(model=final_model))

agent = create_agent(
    model=mini_model,
    tools=[get_location],
    middleware=[change_model]
)

response = agent.invoke({"messages" : [HumanMessage("你知道我的位置吗?")]})
print(response)

可以看到打印结果中的modle_name两次调用的模型是不同的。

1.1.2动态指定工具例子

这个其实也是调下override就ok了。但是需要注意的是，如果刚开始我们的agent中就已经指定好了所有的工具，后面调用时想要根据一定条件筛选工具，仅需要在wrap_model_call中间件进行修改，这样比较简单有效，看一个例子:

from typing import Callable
from langchain.agents import create_agent,AgentState
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from langchain_deepseek import ChatDeepSeek

mini_model = ChatDeepSeek(model="deepseek-v4-flash",base_url="https://api.siliconflow.cn/v1")
model = ChatDeepSeek(model="deepseek-v4-pro",base_url="https://api.siliconflow.cn/v1")

class State(AgentState):
    auth:bool #权限字段，为真时可以获取用户信息

@tool
def get_location() :
    '''获取用户位置的函数'''
    return "北京"

@wrap_model_call(state_schema=State)
def change_model(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse]
)-> ModelResponse | None:
    if not request.state.get("auth",False):
        return handler(request.override(tools=[]))
    return handler(request)


agent = create_agent(
    model=mini_model,
    tools=[get_location],
    middleware=[change_model]
)

response = agent.invoke({
    "messages" : [HumanMessage("你知道我的位置吗?")],
    "auth" : True
})
print(response["messages"][-1].pretty_print())

这样字权限字段设置为False的时候ai就不知道我们的位置了。但是换一种情况，如果最开始我们并没有指定工具，而是需要在中途去添加工具(这个工具我们最开始没有设置到agent中)，那么就需要两个中间件同时配合，在告诉模型有这个工具，然后在实际去调用工具，我们看一个例子:

from typing import Callable
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse, wrap_tool_call
from langchain_core.messages import HumanMessage, ToolMessage
from langchain_core.tools import tool
from langchain_deepseek import ChatDeepSeek
from langgraph.types import Command
from langgraph.prebuilt.tool_node import ToolCallRequest

mini_model = ChatDeepSeek(model="deepseek-v4-flash",base_url="https://api.siliconflow.cn/v1")

@tool
def get_location() :
    '''获取用户位置的函数'''
    return "北京"

@wrap_model_call
def change_model(
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse]
)-> ModelResponse | None:
    #告诉模型有这个工具
    return handler(request.override(tools=[*request.tools,get_location]))

@wrap_tool_call
def change_tool(
    request: ToolCallRequest,
    handler: Callable[[ToolCallRequest], ToolMessage | Command]
) -> ToolMessage | Command | None:
    #实际去添加工具
    if request.tool_call["name"] == "get_location":
        return handler(request.override(tool=get_location))
    return handler(request)

agent = create_agent(
    model=mini_model,
    middleware=[change_model,change_tool]
)

response = agent.invoke({
    "messages" : [HumanMessage("你知道我的位置吗?")],
})
print(response["messages"][-1].content)

所以对于动态添加模型的情况需要注意。

1.1.3动态指定提示词例子

https://reference.langchain.com/python/langchain/agents/middleware/types/dynamic_prompt
这里需要用到一个新的中间件装饰器@dynamic_prompt，看完例子就明白怎么使用了:

from typing import TypedDict
from langchain.agents import create_agent
from langchain.agents.middleware import dynamic_prompt, ModelRequest
from langchain_core.messages import HumanMessage
from langchain_deepseek import ChatDeepSeek

mini_model = ChatDeepSeek(model="deepseek-v4-flash",base_url="https://api.siliconflow.cn/v1")

class Context(TypedDict):
    identify : str

@dynamic_prompt
def change_prompt(
    request : ModelRequest
)-> str:
    Identify = request.runtime.context.get("identify","初学者")
    if Identify == "初学者":
        return "你是一个技术专家，请以简单易懂的话解释用户提到的问题，不要使用专业术语"
    else:
        return "你是一个技术专家，请尽可能的详细的解释用户提到的问题，尽可能的使用专业术语"

agent = create_agent(
    model=mini_model,
    middleware=[change_prompt],
    context_schema=Context
)

response = agent.invoke(
    {"messages" : [HumanMessage("可以简单的介绍下MCP是什么吗?")],},
    context={"identify" : "初学者"}
)
print(response["messages"][-1].content)

输入不同的上下文时，ai给我们反馈的信息的详细程度也不同。

1.2结构化输出

这里和我们之前LangChain的使用方式上大差不差，我们主要看下它相比旧版本的迭代是什么样子的，有兴趣的读者可以自己测试用下，效果和之前也是一样的:

LangChain 通过 create_agent 中的 response_format 参数提供了实现此功能的策略。核心步骤：

  1. 使用 Pydantic 定义期望的输出格式（BaseModel）。
  2. 在创建 Agent 时，将 response_format 参数设置为对应的策略，并传入定义好的格式模型。

1.2.1旧版本策略说明

LangChain 提供了两种主要策略，可以根据模型的支持情况和具体需求进行选择。

策略 原理 适用场景 特点
ToolStrategy（工具策略） 利用模型的工具调用（Tool Calling） 能力，通过创建一个“虚拟工具”来迫使模型以调用该工具参数的形式输出结构化数据。 任何支持工具调用的模型。当模型不支持原生结构化输出或原生输出不可靠时使用。 通用性强，兼容性好。
ProviderStrategy（提供者策略） 直接使用模型提供商提供的原生结构化输出功能。 仅限支持原生结构化输出的模型（如 GPT-4o、Claude 3 等）。 更可靠、效率更高，是首选方案。

1.2.2新版本简化说明

从 LangChain 1.0 开始，提供了一个便捷的简化写法。可以直接将定义好的 Pydantic 模型传给 response_format。

# 简化写法：直接传递模型
agent = create_agent(
    model="gpt-4.1",
    response_format=ContactInfo  # 直接传入模型，而不是策略对象
)

默认行为：

当直接传入模型（如 ContactInfo）时，LangChain 会自动处理：
a. 优先尝试使用 ProviderStrategy（如果模型支持原生结构化输出）。
b. 若不支持，则自动回退使用 ToolStrategy。

这种**“自动选择”**的策略兼顾了便捷性和兼容性。

重要注意事项：

预绑定工具（pre-bound）的模型不支持与结构化输出一起使用。如果需要动态模型选择与结构化输出结合，请确保传入中间件的模型没有预先调用 bind_tools。

1.3自定义状态结构

首先需要特别说明的一点是，从 LangChain 1.0 开始，自定义状态必须是 TypedDict 类型，不再支持 Pydantic 模型或 dataclass。
然后是对于agnet的自定义状态当然必须要继承自AgentState，一般有两种自定义方式，一种是在中间件中指定自定义状态，这也是官方推荐的做法，我们前文也提到过。而另一种则是直接在create_agent处传入自定义状态的state_schema。都很简单有兴趣的读者可以自行下去写一下，这里不再浪费篇幅了。

1.4 Agent中的中断操作

1.4.2与LangGraph中断的区别

首先，因为agent是基于LangGraph构建的，所以它的能力是基于checkpoint的，使用时必须传入checkpoint才可以使用。
其次，Agent中的中断操作并不像LangGraph那样灵活，它只能在工具调用之前进行中断

当 Agent 触发中断时，人工可做出以下三种响应（由策略配置决定哪些可用）：

决策类型 说明 示例用例
approve 完全批准，工具按原参数执行 发送邮件草稿
edit 允许修改工具参数后再执行 修改邮件收件人后发送
reject 拒绝执行，将反馈添加到对话中 拒绝 SQL 删除操作并说明原因

注意：当多个工具调用同时中断时，需按顺序逐一决策；编辑参数时请保守修改，避免影响模型后续判断。

1.4.1使用中断

https://reference.langchain.com/python/langchain/agents/middleware/human_in_the_loop/HumanInTheLoopMiddleware
使用 HumanInTheLoopMiddleware 中间件完成人机协作，其配置选项主要包含以下两部分：

参数一：interrupt_on（必选）

一个字典，用于定义哪些工具需要触发人工中断以及允许哪些决策类型。

键为工具名称（字符串），值可以是以下三种形式：

值类型 含义 示例
True 中断该工具，允许全部三种决策（approve、edit、reject） "工具1": True
False 不中断，自动批准（工具调用直接执行） "工具2": False
InterruptOnConfig 对象 精细控制：可指定允许的决策列表和自定义描述 见下方

InterruptOnConfig 对象的属性：

  • allowed_decisions：列表，可选值 "approve"、"edit"、"reject"。决定人工可用的操作类型。
  • description：字符串或可调用函数，用于覆盖该工具的中断提示消息。若未指定，则使用全局 description_prefix 拼接而成。
interrupt_on={
    "write_file": True, # 全决策可用
    "execute_sql": {"allowed_decisions": ["approve", "reject"]}, # 禁止编辑
    "read_data": False, # 自动通过
}

参数二：description_prefix（可选）

字符串。作为中断消息的全局前缀，默认会在每个中断请求的描述前加上此文本。

最终消息格式：description_prefix + "\n\nTool: <tool_name>\nArgs: <arguments>"

工具级覆盖：若在 InterruptOnConfig 中指定了 description，则忽略全局前缀。

示例：

description_prefix="工具执行尚待批准"

中断时用户看到的消息开头为：

工具执行尚待批准
Tool: execute_sql
Args: {…}

我们看一个例子:

from typing import TypedDict
from langchain.agents import create_agent
from langchain.agents.middleware import dynamic_prompt, ModelRequest, HumanInTheLoopMiddleware
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from langchain_deepseek import ChatDeepSeek
from langgraph.checkpoint.memory import InMemorySaver

mini_model = ChatDeepSeek(model="deepseek-v4-flash",base_url="https://api.siliconflow.cn/v1")

@tool
def get_weather(query : str) -> str | None :
    '''生成二维码的工具函数
        Args:
            query (str):用户想要查询的目标位置
        Returns:
            str:成功返回对应位置天气信息总结，失败返回None
    '''
    return f"{query}最近一周天气总是阴雨绵绵"

agent = create_agent(
    model=mini_model,
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "get_weather": True, # 全决策可用与下面的方式相同
                #"get_location": {"allowed_decisions": ["approve", "reject","edit"]}
            },
            description_prefix="工具执行尚待批准"
        )
    ],
    tools=[get_weather],
    checkpointer=InMemorySaver()
)

config = {"configurable" : {"thread_id" : "111"}}

response = agent.invoke(
    {"messages" : [HumanMessage("可以告诉我焦作最近一周的天气是怎样的吗?")],},
    config=config,
    version="v2" #必须使用v2版本获取中断信息
)

#使用interrupts获取中断信息
print(response.interrupts)

输出的中断信息是这样的:

(Interrupt(
    value={
        'action_requests': [{
            'name': 'get_weather',
            'args': {'query': '焦作'},
            'description': "工具执行尚待批准\n\nTool: get_weather\nArgs: {'query': '焦作'}"
        }],
        'review_configs': [{
            'action_name': 'get_weather',
            'allowed_decisions': ['approve', 'edit', 'reject', 'respond']}
        ]},
    id='d885cac899fca998a61c55b6891c0dbe'),
)

我们来分别看下三种策略的中断回复，首先是最简单的批准:

response = agent.invoke(
    Command(
        resume={
            "decisions" : [{
                "type" : "approve"
            }]
        }
    ),
    #以下部分保持不变
    config=config,
    version="v2" #必须使用v2版本获取中断信息
)
#新版本推荐response.value["messages"][-1].content这样访问
#旧版本response["messages"][-1].content字典方式访问已经不再推荐
print(response.value["messages"][-1].content)

然后是编辑:

response = agent.invoke(
    Command(
        resume={
            "decisions" : [{
                "type" : "edit",
                "edited_action": {
                # 要调用的工具名称。
                # 通常会与原始动作保持⼀致。
                "name": "get_weather",
                # 要传递给工具的参数。
                "args":
                    {
                        "query": "北京",
                    },
                }
            }]
        }
    ),
    #以下部分保持不变
    config=config,
    version="v2" #必须使用v2版本获取中断信息
)
#新版本推荐response.value["messages"][-1].content这样访问
#旧版本response["messages"][-1].content字典方式访问已经不再推荐
print(response.value["messages"][-1].content)

最后是拒绝:

response = agent.invoke(
    Command(
        resume={
            "decisions" : [{
                "type" : "reject",
                "message" : "当前用户禁止访问目标工具" #拒绝原因->实际上被包装为一个ToolMessage插入到了状态的messages列表中
            }]
        }
    ),
    #以下部分保持不变
    config=config,
    version="v2" #必须使用v2版本获取中断信息
)
#新版本推荐response.value["messages"][-1].content这样访问
#旧版本response["messages"][-1].content字典方式访问已经不再推荐
print(response.value["messages"][-1].content)

1.5流式输出结果

与LangGraph一样，Agent也支持三种流式输出模式updates，messages与custom，且效果与LangGraph一样。同样的，也支持组合流输出。
可以通过create_agent中的stream_mode参数指定输出模式。

1.5.1新旧版本接收流式输出结果

在agent.stream时，如果没有指定version="v2"，那么采用的是旧版本的流式返回方式，就和我们之前在LangGraph中那样的接收方式一样:

# 必须使用 (mode, data) 元组
for mode, chunk in agent.stream(
    {"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
    stream_mode=["updates", "custom"],
):
    print(mode)   # "updates" or "custom"
    print(chunk)  # payload

在指定的情况下，agent.stream仅返回一个chunk，这个是mode与chunk的组合版本:

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
    stream_mode=["updates", "custom"],
    version="v2",
):
    print(chunk["type"])   # "updates" or "custom"
    print(chunk["data"])   # payload

  • version=“v2”：使用新版流式输出格式（推荐）。与旧版 (mode, data) 元组格式不同，v2 格式返回一个字典，包含 “type” 和 “data” 两个键

  • chunk["type"]：流模式名称（“updates” 或 “custom”）

  • chunk["data"]：该模式对应的实际数据载荷

1.5.2例子:流式输出推理与回答内容

在调用ds系列模型的时候默认就是带上推理的(在测试的时候偶然发现的)，所以我们来看一个例子，让它将正常内容，推理内容，工具返回结果分开进行打印展示:

from langchain.agents import create_agent
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from langchain_deepseek import ChatDeepSeek

mini_model = ChatDeepSeek(model="deepseek-v4-flash",base_url="https://api.siliconflow.cn/v1")

@tool
def get_weather(query : str) -> str | None :
    '''生成二维码的工具函数
        Args:
            query (str):用户想要查询的目标位置
        Returns:
            str:成功返回对应位置天气信息总结，失败返回None
    '''
    return f"{query}最近一周天气总是阴雨绵绵"

agent = create_agent(
    model=mini_model,
    tools=[get_weather],
)

n = 1

for chunk in agent.stream(
    {"messages" : [HumanMessage("可以告诉我焦作最近一周的天气是怎样的吗?")],},
    stream_mode="messages",
    version="v2" #必须使用v2版本获取中断信息
):
    print(chunk)

打印出来的chunk如果是AIMessage，其内容中的content不为空时即正常输出内容，如果为空但是其additional_kwargs中的reasoning_content不为空的时候那么此内容为推理输出的内容，比如:

{'type': 'messages', 'ns': (), 'data': (AIMessageChunk(content='', additional_kwargs={'reasoning_content': '用户'},...

这就是一个推理内容的输出块。所以要想把推理内容和正常回复内容都打印出来，我们可以这样子改下打印策略:

last_chunk_type = "normal"

for chunk in agent.stream(
    {"messages" : [HumanMessage("可以告诉我焦作最近一周的天气是怎样的吗?")],},
    stream_mode="messages",
    version="v2" #必须使用v2版本获取中断信息
):
    if isinstance(chunk["data"][0],ToolMessage):
        print("\n工具调用信息如下:")
        print(chunk["data"][0].content, end="", flush=True)
    elif chunk["data"][0].content:
        if last_chunk_type == "reasoning":
            print("\n正常内容输出如下:")
        print(chunk["data"][0].content,end="",flush=True)
        last_chunk_type = "normal"
    elif chunk["data"][0].additional_kwargs.get("reasoning_content",""):
        if last_chunk_type == "normal":
            print("\n推理内容输出如下:")
        print(chunk["data"][0].additional_kwargs.get("reasoning_content"), end="", flush=True)
        last_chunk_type = "reasoning"

一个结果例子是这样的:

推理内容输出如下:
用户想知道焦作最近一周的天气情况。我需要调用get_weather工具来获取信息。
正常内容输出如下:
好的，我来帮你查询焦作最近一周的天气情况！
工具调用信息如下:
焦作最近一周天气总是阴雨绵绵
推理内容输出如下:
工具返回了结果，但比较简略，说"焦作最近一周天气总是阴雨绵绵"。我需要把这个信息用更友好的方式告诉用户。
正常内容输出如下:
根据查询结果，焦作最近一周的天气情况如下：

🌧️ **焦作未来一周天气趋势：**

最近一周焦作**以阴雨天气为主**，总是阴雨绵绵。建议你：

- **随身携带雨伞**，因为降雨会比较频繁
- **注意添衣保暖**，阴雨天体感温度可能会偏低
- **出行注意安全**，雨天路滑，驾车或步行都要多加小心

如果你需要了解更具体的每日天气详情（如温度、降雨量等），也可以告诉我，我可以进一步帮你查询！😊

如果是OPENAI系列的模型，我们来看一个例子，来分开进行流式的输出:

from langchain.agents import create_agent
from langchain.messages import AIMessageChunk
from langchain_openai import ChatOpenAI

# 1. 配置模型，启用推理输出
model = ChatOpenAI(
    model="gpt-5-mini",
    reasoning={                    # 关键配置：
        "effort": "medium",        # 推理程度：'low', 'medium', or 'high'
        "summary": "auto",         # 推理摘要：'detailed', 'auto', or None
    }
)

def get_weather(city: str) -> str:
    """获取天气"""
    return f"{city} 天气晴朗！"

agent = create_agent(model=model, tools=[get_weather])

# 2. 使用 stream_mode="messages" 并过滤 reasoning 块
for token, metadata in agent.stream(
    {"messages": [{"role": "user", "content": "上海的天气如何？"}]},
    stream_mode="messages",
):
    if not isinstance(token, AIMessageChunk):
        continue

    # 推理内容
    reasoning = [b for b in token.content_blocks if b["type"] == "reasoning"]
    # 响应内容
    text = [b for b in token.content_blocks if b["type"] == "text"]

    if reasoning and 'reasoning' in reasoning[0]:
        print(f"{reasoning[0]['reasoning']}", end="")
    if text:
        print(text[0]["text"], end="")

(这里就不给结果演示了，因为作者一直使用的是中间站调用openai系列模型，但是写文章的时候挂掉了，所以展示不了了喵T_T)。

当然复杂的组合模式和LangGraph那里演示的一样，这里就不再多展示了。

二.MCP认识及快速上手

2.1什么是MCP

https://modelcontextprotocol.io/docs/getting-started/intro
模型上下文协议 (Model Context Protocol，简称MCP)，它是干什么用的呢。文档中有详细的概念的叙述，我们就不那么详细了，这里就直白些。

像我们之前去编写我们的LangChain代码时，每次新开一个项目时就需要把原来的工具重新写一遍，非常麻烦(可以复制粘贴也很麻烦)。
有没有什么办法，我们直接部署一个服务，然后让本地的LangChain应用通过我们给定的服务器地址调用服务，像调工具那样。当然可以，但是agent应该以什么样的方式去向服务器发起调用呢?
就像我们的网页通常以发起HTTP请求来规范我们访问服务器上部署的后端服务一样，agent想要调用后端部署的服务，也需要遵循事先与服务端约定好的协议。这个协议就是MCP。

下面是关于MCP的几点补充:
1. 它不是“HTTP之于网页”那样的通用协议
实际上，MCP 是基于 JSON-RPC 2.0 构建的，可以理解为“协议之上的协议”。即便你完全不用 MCP，用 HTTP + 自定义 JSON 格式也能实现 Agent 调用远程服务。MCP 的价值在于它定义了“格式”标准——比如工具请求长什么样、返回结果长什么样、错误怎么处理，而不是重新发明传输层。

2. 它不只解决工具复用问题
复用 LangChain 工具”这一场景上，这确实是痛点之一。但 MCP 的野心更大：

  • 不只管工具，还管数据和 Prompt：MCP 定义的三大核心原语是 Resources（数据源）、Tools（工具） 和 Prompts（预定义模板）。也就是说，AI 应用不仅可以通过 MCP 调用远程计算，还能读取远程文件、数据库，甚至获取预设的 Prompt 模板。

  • 不只解决复用问题：如官方文档所言，MCP 的核心是标准化连接——让 AI 应用“一次接入，到处使用”，像 USB-C 一样。你的复用问题只是这个标准化价值的体现之一。

3. 它不需要“事先约定”，而是“动态发现”

虽然说Agent 需要“遵循事先约定好的协议”来调用服务确实没什么问题。但是更准确地说，MCP 是动态发现的：Agent 启动时连接 MCP Server，Server 会主动告诉 Agent“我提供了什么工具和数据”，Agent 动态发现并使用。

4. 它不是专为 Agent 设计的

虽然 Agent 是 MCP 的主要应用场景，但 MCP 的官方定义是“连接 AI 应用到外部系统的标准”。即便是没有 Agent 能力的普通聊天应用，也可以通过 MCP 获取工具和数据。当然，目前主流应用确实是 Agent。

2.2两种MCP传输方式

MCP 支持两种主要的客户端-服务器通信传输机制。

2.2.1HTTP（也称 streamable-http）

  • 通过 HTTP 请求通信。有关详细信息，请参见 MCP HTTP 运输规范。
  • 适合远程服务器、云部署。
  • 支持传递自定义请求头（如认证 token）和实现 httpx.Auth 接口的认证机制。
  • 官方示例:自定义身份验证实现

2.2.2stdio

  • 客户端将服务器作为子进程启动，通过标准输入/输出通信。
  • 适合本地工具、简单配置。
  • 有状态特性：子进程在客户端连接期间持续存在，但 MultiServerMCPClient 默认仍为每次工具调用创建新会话。

具体例子可以参考下面的快速上手部分。

2.3快速上手

2.3.1定义服务端

因为LangChain仅提供了客户端的MCP工具，所以我们需要借助其他的三方包去实现MCP服务端，这里使用fast-mcp，安装包的方式如下:

pip install fastmcp

我们分别定义两个不同传输方式的mcp服务器并运行起来:
stdio:

from fastmcp import FastMCP

mcp = FastMCP("Weather")

@mcp.tool()
def get_weather_by_location(loc : str):
    """在已知用户位置的情况下查询天气信息"""
    return f"{loc}未来一个月的天气均是阳光明媚"

@mcp.tool()
def get_weather_without_location():
    """未知用户未知的情况下查询天气信息"""
    return "焦作的未来一个月的天气均是阳光明媚"

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

streamable-http:

from fastmcp import FastMCP

mcp = FastMCP("Math")

@mcp.tool()
def add(a : int,b : int):
    """求两数之和"""
    return a + b

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

2.3.2定义客户端

接下来来定义客户端:
使用 langchain-mcp-adapters 库让 LangChain Agent 调用 MCP 服务器上定义的工具。

pip install langchain-mcp-adapters

核心用法：

  • 创建 MultiServerMCPClient，配置一个或多个 MCP 服务器（支持 stdio 和 http 传输）。
  • 调用 client.get_tools() 获取所有工具。
  • 将工具传入 create_agent，构建 Agent。

我们来创建一个客户端链接我们刚才创建好的两个mcp服务:

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

# FastMCP 客户端是异步的，因此我们需要使用 asyncio.run 来运行客户端
async def main():
    client = MultiServerMCPClient(
        {
            "Weather": {
                "transport": "stdio",  # 本地子进程通信
                "command": "python",
                # stdio服务源码文件的绝对路径
                "args": ["C:/code/LangChain/LangChain-v1-test/studio-mcp-test.py"],
            },
            "Math": {
                "transport": "streamable-http",  # 基于 HTTP 的远程服务器
                # Math服务的启动地址
                "url": "http://127.0.0.1:8080/mcp",
            }
        }
    )
    #获取工具并实例化agent
    tools = await client.get_tools()
    print(tools)
    model = ChatDeepSeek(model="deepseek-v4-flash", base_url="https://api.siliconflow.cn/v1")
    agent = create_agent(
        model=model,
        tools=tools,
        system_prompt="你是一个善于调用工具回答用户问题的助手，回答用户问题前，如果有合适的工具调用，请务必调用工具作为辅助回答用户的问题。"
    )

    math_response = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "3 + 5等于多少?"}]}
    )

    weather_response = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "上海的天气怎么样？"}]}
    )

    print(math_response)
    print(weather_response)


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

将客户端启动便可以看到有哪些工具可供调用以及两个问题的回答信息了。

补充注意事项：

  • FastMCP 客户端是异步的，因此我们需要使用 asyncio.run 来运行客户端。

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

2.3.3补充:有状态会话

如果需要控制 MCP 会话的生命周期，例如处理维护跨工具调用上下文的有状态服务器时，可以使用 client.session() 创建持久的 ClientSession。使用 client.session("服务器名称") 上下文管理器，配合 load_mcp_tools 加载工具。
基于上面的例子我们来进行修改，让它称为一个有状态的会话:

import asyncio
from langchain_deepseek import ChatDeepSeek
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
from langchain_mcp_adapters.tools import load_mcp_tools


# FastMCP 客户端是异步的，因此我们需要使用 asyncio.run 来运行客户端
async def main():
    client = MultiServerMCPClient(
        {
            "Weather": {
                "transport": "stdio",  # 本地子进程通信
                "command": "python",
                # stdio服务源码文件的绝对路径
                "args": ["C:/code/LangChain/LangChain-v1-test/studio-mcp-test.py"],
            },
            "Math": {
                "transport": "streamable-http",  # 基于 HTTP 的远程服务器
                # Math服务的启动地址
                "url": "http://127.0.0.1:8080/mcp",
            }
        }
    )
    async with client.session("Weather") as session_weather, client.session("Math") as session_math:
        tools_weather = await load_mcp_tools(session_weather)
        tools_math = await load_mcp_tools(session_math)
        tools = tools_weather + tools_math
        print(tools)
        # 该 agent 的所有工具调用将复用同⼀个会话
        model = ChatDeepSeek(model="deepseek-v4-flash", base_url="https://api.siliconflow.cn/v1")
        agent = create_agent(
            model=model,
            tools=tools,
            system_prompt="你是一个善于调用工具回答用户问题的助手，回答用户问题前，如果有合适的工具调用，请务必调用工具作为辅助回答用户的问题。"
        )

        math_response = await agent.ainvoke(
            {"messages": [{"role": "user", "content": "3 + 5等于多少?"}]}
        )

        weather_response = await agent.ainvoke(
            {"messages": [{"role": "user", "content": "上海的天气怎么样？"}]}
        )

        print(math_response)
        print(weather_response)


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

这样子去创建会话那么对于这所有的mcp服务器的请求此时都不会再创建新会话了。

# 数据库 # python
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

WalkingLab 组织入驻 AtomGit，释放高校智能体开源科研成果

avatar AtomGit开源社区
cover

拼手速！GLM-5.2免费Token每天10点准点开抢！

avatar AtomGit开源社区
cover

圆满落幕｜智驱迭代・昇腾赋能 AI Agent 行业实践 Meetup 上海站，全栈落地干货一次吃透

avatar AtomGit开源社区