LangGraph MCP集成：机器人工具调用入门

LangGraph工具调用基础

话说LLM在处理实时信息、数学计算这类问题上，表现一直不太理想。所以一个合格的Agent，得会调用外部工具来补短板。这篇就从最常见的Web搜索工具入手，带着大家快速上手LangGraph的工具调用。这里拿Ta vilySearch（app.ta vily.com）做演示——新用户有1000次免费搜索额度，够用一阵了。

在之前构建聊天机器人的基础上，只需要加少量代码就能完成工具调用。LangGraph已经把大部分复杂逻辑给封装好了，省心不少。

示例代码

fromtypingimportAnnotated
fromtyping_extensionsimportTypedDict
fromlangchain_openaiimportChatOpenAI
fromlanggraph.graphimportStateGraph, START
fromlanggraph.graph.messageimportadd_messages
frompydanticimportSecretStr
fromlangchain_ta vilyimportTa vilySearch
fromlanggraph.prebuiltimportToolNode, tools_condition

# 此处定义你自己的模型
llm = ChatOpenAI(base_url="http://127.0.0.1:8000/v1", api_key=SecretStr("123123"), model="qwen3_32")

# 定义工具
tool = Ta vilySearch(ta vily_api_key="你的ta vily apikey", max_results=2)
tools = [tool]

# 工具绑定到模型
llm_with_tools = llm.bind_tools(tools)

# 定义图状态
classState(TypedDict):
    messages: Annotated[list, add_messages] # 此处维护完整的消息历史

graph = StateGraph(State)

defchatbot(state: State):
    return{"messages": [llm_with_tools.invoke(state["messages"])]}

# 使用LangGraph提供的工具节点
tool_node = ToolNode(tools=tools)

graph.add_node("chatbot", chatbot)
# 添加工具节点
graph.add_node("tools", tool_node)

# 添加工具条件分支
graph.add_conditional_edges(
    "chatbot",
    tools_condition,
)

graph.add_edge("tools","chatbot")
graph.add_edge(START,"chatbot")

app = graph.compile()

if__name__ =="__main__":
    messages = []
    whileTrue:
        user_input =input("?‍?: ")
        ifuser_input.lower()in["quit","exit","q"]:
            print("Exiting...")
            break
        messages.append({"role":"user","content": user_input})
        response = app.invoke({"messages": messages})
        messages = response["messages"]
        print(f'?:{response["messages"][-1].content}')

效果如下

LangGraph自定义工具与工具节点

LangChain预置了不少常见工具，比如搜索类的（Bing、SerpAPI、Ta vily），代码执行类的（Python REPL、Node.js REPL），还有数据库类、Web数据类、天气API等等。但说实话，真到了企业开发场景，更常见的是自己写定制工具——毕竟每个业务都有自己的特殊需求。

下面演示一个基于BaseTool的自定义Ta vily搜索工具，以及怎么在LangGraph里接进去。

示例代码

importjson
fromlangchain_core.messagesimportToolMessage
fromlanggraph.constantsimportEND
fromtypingimportAnnotated,Type,Optional
fromlangchain_core.callbacksimportCallbackManagerForToolRun, AsyncCallbackManagerForToolRun
fromtyping_extensionsimportTypedDict
fromlangchain_openaiimportChatOpenAI
fromlanggraph.graphimportStateGraph, START
fromlanggraph.graph.messageimportadd_messages
frompydanticimportSecretStr, BaseModel, Field
fromlangchain_core.toolsimportBaseTool
fromta vilyimportTa vilyClient, AsyncTa vilyClient

# 自定义工具部分
classTa vilySearchInput(BaseModel):
    query:str= Field(description=("搜索查询"))

classTa vilySearchTool(BaseTool):
    name:str="ta vily_search"
    description:str="""一个针对全面、准确和可信的结果进行了优化的搜索引擎。当需要回答有关时事的问题时很有用。输入应该是搜索查询。"""
    args_schema:Type[BaseModel] = Ta vilySearchInput
    # return_direct: bool = True
    def_run(self, query:str, run_manager:Optional[CallbackManagerForToolRun] =None ) ->int:
        client = Ta vilyClient()
        search_r = client.search(query=query, max_results=2)
        returnsearch_r

    asyncdef_arun(self,query:str,run_manager:Optional[AsyncCallbackManagerForToolRun] =None, ) ->int:
        client = AsyncTa vilyClient()
        search_r =awaitclient.search(query=query, max_results=2)
        returnsearch_r

# 自定义的工具执行节点
classToolNode:
    def__init__(self, tools:list) ->None:
        self.tools_by_name = {tool.name: toolfortoolintools}
    def__call__(self, inputs:dict):
        ifmessages := inputs.get("messages", []):
            message = messages[-1]
        else:
            raiseValueError("No message found in input")

        outputs = []
        fortool_callinmessage.tool_calls:
            print(f'正在执行工具{tool_call["name"]}，参数{tool_call["args"]}')
            tool_result =self.tools_by_name[tool_call["name"]].invoke(tool_call["args"])
            print(f'工具{tool_call["name"]}, 执行结果{json.dumps(tool_result, ensure_ascii=False)}')
            outputs.append(ToolMessage(content=json.dumps(tool_result, ensure_ascii=False),name=tool_call["name"],tool_call_id=tool_call["id"],))
        return{"messages": outputs}

# 此处定义你自己的模型
llm = ChatOpenAI(base_url="http://127.0.0.1:8000/v1", api_key=SecretStr("123123"), model="qwen3_32")

# 定义工具
tool = Ta vilySearchTool()
tools = [tool]

# 工具绑定到模型
llm_with_tools = llm.bind_tools(tools)

# 定义图状态
classState(TypedDict):
    messages: Annotated[list, add_messages] # 此处维护完整的消息历史

graph = StateGraph(State)

defchatbot(state: State):
    return{"messages": [llm_with_tools.invoke(state["messages"])]}

# 工具路由
defroute_tools(state: State):
    ifisinstance(state,list):
        ai_message = state[-1]
    elifmessages := state.get("messages", []):
        ai_message = messages[-1]
    else:
        raiseValueError(f"No messages found in input state to tool_edge:{state}")
    ifhasattr(ai_message,"tool_calls")andlen(ai_message.tool_calls) >0:
        return"tools"
    returnEND

# 使用LangGraph提供的工具节点
tool_node = ToolNode(tools=tools)

graph.add_node("chatbot", chatbot)
# 添加工具节点
graph.add_node("tools", tool_node)

# 添加工具条件边
graph.add_conditional_edges(
    "chatbot",
    route_tools,
    {"tools":"tools", END: END},
)

graph.add_edge("tools","chatbot")
graph.add_edge(START,"chatbot")

app = graph.compile()

if__name__ =="__main__":
    messages = []
    whileTrue:
        user_input =input("?‍?: ")
        ifuser_input.lower()in["quit","exit","q"]:
            print("Exiting...")
            break
        messages.append({"role":"user","content": user_input})
        response = app.invoke({"messages": messages})
        messages = response["messages"]
        print(f'?:{response["messages"][-1].content}')

效果如下

除了继承BaseTool，LangChain还支持通过@tool装饰器来快速定义工具。这种方法更简洁，适合轻量级场景。

fromtypingimportAnnotated
fromta vilyimportTa vilyClient
fromlangchain_core.toolsimporttool

@tool("ta vily_search")
defta vily_search_tool(query: Annotated[str,"搜索查询"]):
    """一个针对全面、准确和可信的结果进行了优化的搜索引擎。当需要回答有关时事的问题时很有用。输入应该是搜索查询。"""
    client = Ta vilyClient()
    search_r = client.search(query=query)
    returnsearch_r

ta vily_search_tool.invoke({"query":"北京2025年8月27日天气怎么样？"})

LangGraph调用MCP

除了自定义工具，LangGraph还支持MCP（Model Context Protocol），这让工具的复用和扩展性更强。要用MCP，需要额外安装一个包：

pip install langchain-mcp-adapters

编写一个简单的MCP Server

还是以Ta vily搜索为例：

importjson
frommcp.server.fastmcpimportFastMCP
fromta vilyimportTa vilyClient

mcp = FastMCP("search")

@mcp.tool()
asyncdefta vily_search(query:str) ->str:
    """一个针对全面、准确和可信的结果进行了优化的搜索引擎。当需要回答有关时事的问题时很有用。输入应该是搜索查询。"""
    client = Ta vilyClient()
    search_r = client.search(query=query, max_results=2)
    returnjson.dumps(search_r, ensure_ascii=False)

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

LangGraph+MCP完整示例代码

importasyncio
fromlangchain_mcp_adapters.clientimportMultiServerMCPClient
fromlanggraph.prebuiltimportToolNode, tools_condition
fromtypingimportAnnotated
fromtyping_extensionsimportTypedDict
fromlangchain_openaiimportChatOpenAI
fromlanggraph.graphimportStateGraph, START
fromlanggraph.graph.messageimportadd_messages
frompydanticimportSecretStr

# 此处定义你自己的模型
llm = ChatOpenAI(base_url="http://127.0.0.1:8000/v1", api_key=SecretStr("123123"), model="qwen3_32")

# 配置MCP Server
client = MultiServerMCPClient({
    "search": {
        "url":"http://localhost:8000/mcp/",
        "transport":"streamable_http",
    }
})

classState(TypedDict):
    messages: Annotated[list, add_messages] # 此处维护完整的消息历史

graph = StateGraph(State)

asyncdefmain():
    tools =awaitclient.get_tools()

    # 工具绑定到模型
    llm_with_tools = llm.bind_tools(tools)

    defchatbot(state: State):
        return{"messages": [llm_with_tools.invoke(state["messages"])]}

    graph = StateGraph(State)
    graph.add_node(chatbot)
    graph.add_node(ToolNode(tools))
    graph.add_edge(START,"chatbot")
    graph.add_conditional_edges(
        "chatbot",
        tools_condition,
    )
    graph.add_edge("tools","chatbot")
    app = graph.compile()

    messages = []
    whileTrue:
        user_input =input("?‍?: ")
        ifuser_input.lower()in["quit","exit","q"]:
            print("Exiting...")
            break
        messages.append({"role":"user","content": user_input})
        response =awaitapp.ainvoke({"messages": messages})
        messages = response["messages"]
        print(f'?:{response["messages"][-1].content}')

asyncio.run(main())

效果如下

以上所有的LangGraph图结构

学习资源推荐

如果你想更深入地学习大模型，以下是一些非常有价值的学习资源，这些资源将帮助你从不同角度学习大模型，提升你的实践能力。

来源：互联网