工具

之前的章节中，我们构建的智能体曾多次使用过“工具（Tool）”，例如第一章Demo里的get_weather函数。工具（Tool）是智能体与外部世界交互的核心机制，智能体需要通过工具调用完成搜索、查询数据库、执行代码、调用API等各种操作。这篇笔记我们将系统介绍如何定义工具、如何在普通ChatModel和智能体中使用工具，以及如何通过MCP协议接入海量的第三方工具。

工具调用的本质

LLM为什么能调用工具？实际上没什么神奇的，调用工具对于LLM来说仍是输入一段文本，推理后得到一些响应文本，只不过在工具调用时，我们在提示词里描述了都有哪些可用工具、参数类型、返回值类型、用途等，LLM会按照规则输出想要调用的函数和参数值，我们解析出这些信息后，真正去反射调用工具函数的仍是代码。

在LangChain中，工具（Tool）本质上就是一个带有元数据的Python函数。LangChain将这个函数包装成一个BaseTool对象，并附带名称（name）、描述（description）和参数Schema（args_schema）信息，这些元数据最终会发送给LLM，LLM根据这些信息判断何时、以何种参数调用该工具。

定义工具

使用@tool装饰器

最简单也是最推荐的工具定义方式是使用@tool装饰器，下面是一个例子。

from langchain_core.tools import tool


@tool
def get_weather(city: str) -> str:
    """Get current weather for a given city.

    Args:
        city: City name, for example "Los Angeles" or "Beijing".

    Returns:
        A weather summary string, for example "Sunny 14°C".
    """
    # 实际项目中这里应调用真实的天气API
    return "晴 14℃"

工具函数一定要有参数类型约束和docstring注释，编写高质量的docstring非常重要，尤其是多工具或工具的用途比较复杂，仅靠函数名难以判断的时候，这些信息对LLM选择工具和正确调用工具非常有帮助。装饰器@tool会自动从函数名和docstring中提取工具的名称和描述，并根据类型注解生成参数Schema。定义完成后，我们可以查看工具的元信息。

print(get_weather.name)        # get_weather
print(get_weather.description) # Get current weather for a given city. ...
print(get_weather.args)        # {'city': {'title': 'City', 'type': 'string', ...}}

多参数工具

工具可以定义多个参数，LangChain会自动根据类型注解生成对应的参数Schema。

from langchain_core.tools import tool


@tool
def search_product(product_name: str, category: str = "all", max_results: int = 10) -> list[dict]:
    """Search for products in the e-commerce database.

    Args:
        product_name: The name or keywords to search for.
        category: Product category filter, e.g. "electronics", "clothing". Defaults to "all".
        max_results: Maximum number of results to return. Defaults to 10.

    Returns:
        A list of product dicts with name, price, and category fields.
    """
    # 模拟返回结果
    return [{"name": product_name, "price": 99.9, "category": category}]

使用Pydantic定义参数Schema

对于复杂的参数结构，可以使用Pydantic模型显式定义参数Schema，让参数约束更加精确清晰。

from langchain_core.tools import tool
from pydantic import BaseModel, Field


class SendEmailInput(BaseModel):
    to: str = Field(description="Recipient email address")
    subject: str = Field(description="Email subject line, keep it concise")
    body: str = Field(description="Email body content in plain text or markdown")
    cc: list[str] = Field(default=[], description="CC email addresses, optional")


@tool(args_schema=SendEmailInput)
def send_email(to: str, subject: str, body: str, cc: list) -> str:
    """Email the specified recipient.

    Returns:
        Confirmation message indicating success or failure.
    """
    if cc is None:
        cc = []
    recipients = [to] + cc
    return f"Email sent to {', '.join(recipients)} with subject '{subject}'"

继承BaseTool实现工具类

除了使用@tool装饰器，LangChain中定义工具还有一种更灵活的方式是继承BaseTool实现工具类，它适合需要维护内部状态或依赖注入的场景。

from langchain_core.tools import BaseTool
from pydantic import Field


class DatabaseQueryTool(BaseTool):
    name: str = "database_query"
    description: str = "Query the database with SQL and return results as a string."

    # 可以在工具类中定义依赖项
    connection_string: str = Field(default="sqlite:///app.db")

    def _run(self, query: str) -> str:
        """Execute a SQL query and return the results."""
        # 实际项目中这里连接数据库并执行查询
        return f"Query '{query}' executed successfully. 3 rows returned."

    async def _arun(self, query: str) -> str:
        """Async version of _run."""
        return self._run(query)

在普通ChatModel中使用工具

绑定工具

基础ChatModel本身不具备工具调用能力，但我们可以通过bind_tools()方法将工具列表绑定到模型上，此后模型在输出时可能会包含工具调用请求（而非直接输出文本）。

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from datetime import datetime
import pytz


@tool
def get_weather(city: str) -> str:
    """Get current weather for a given city."""
    return "晴 14℃"


@tool
def get_time(timezone: str) -> str:
    """Get the current time in a given timezone, e.g. 'Asia/Shanghai'."""
    tz = pytz.timezone(timezone)
    return datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S")


model = ChatOpenAI(
    model="qwen3:30b-a3b",
    base_url="http://localhost:11434/v1/",
    api_key="dummy",
    temperature=1,
    top_p=1,
    max_tokens=16384,
    timeout=120,
    max_retries=6
)

# 绑定工具列表到模型
model_with_tools = model.bind_tools([get_weather, get_time])

# 调用绑定了工具的模型
resp = model_with_tools.invoke("洛杉矶现在的天气怎么样？")
print(resp.content)  # 如果LLM决定调用工具，这里可能为空
print(resp.tool_calls)  # 工具调用请求列表

执行上面代码后，一般来说，模型会输出工具调用请求。

[{'name': 'get_weather', 'args': {'city': 'Los Angeles'}, 'id': 'call_72099e1c761e45c984acea', 'type': 'tool_call'}]

手动处理工具调用

使用bind_tools()后，模型返回了工具调用请求，我们需要手动处理这些请求、执行工具、并将结果反馈给模型。

from langchain_core.messages import HumanMessage, ToolMessage

# 第一轮：用户提问，模型决定调用工具
messages = [HumanMessage("洛杉矶现在天气如何？")]
resp = model_with_tools.invoke(messages)
messages.append(resp)  # 将AI的工具调用请求加入消息历史

# 执行工具调用
for tool_call in resp.tool_calls:
    tool_name = tool_call["name"]
    tool_args = tool_call["args"]

    # 根据工具名称找到对应工具并执行
    tools_map = {"get_weather": get_weather, "get_time": get_time}
    tool_result = tools_map[tool_name].invoke(tool_args)

    # 将工具结果以 ToolMessage 的形式加入消息历史
    messages.append(ToolMessage(
        content=str(tool_result),
        tool_call_id=tool_call["id"]
    ))

# 第二轮：将工具结果发回给模型，得到最终回答
final_resp = model_with_tools.invoke(messages)
print(final_resp.content)

上面的手动处理流程比较繁琐，实际开发中我们基本不会这样写，但这也正是智能体（Agent）存在的意义，智能体封装了这个循环，让我们不必手动处理工具调用的循环逻辑。

在智能体中使用工具

智能体会自动处理工具调用的完整循环，我们只需将工具列表传给create_agent()即可。这部分内容在"构建智能体"章节中已有详细介绍，这里简要回顾。

from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI


@tool
def get_weather(city: str) -> str:
    """Get current weather for a given city."""
    return "晴 14℃"


@tool
def get_population(city: str) -> str:
    """Get the current population of a given city."""
    return "约400万人"


model = ChatOpenAI(
    model="qwen3:30b-a3b",
    base_url="http://localhost:11434/v1/",
    api_key="dummy",
    temperature=1,
    top_p=1,
    max_tokens=16384,
    timeout=120,
    max_retries=6
)

agent = create_agent(
    model=model,
    tools=[get_weather, get_population],
    system_prompt="You are a helpful AI assistant that can answer questions about cities.",
)

resp = agent.invoke({
    "messages": [{"role": "user", "content": "洛杉矶的天气和人口分别是多少？"}]
})
print(resp["messages"][-1].content)

智能体会自动决定调用哪些工具、以什么顺序调用、调用多少次，最终综合所有工具的结果生成回答。

接入MCP工具

什么是MCP

MCP（Model Context Protocol，模型上下文协议）是由Anthropic提出的一种开放标准，它试图统一AI应用与外部工具、数据源之间的连接方式，这有点类似于AI领域的“USB”接口标准，相比于前面介绍的内置工具，MCP更像是连接到我们应用中的“外设”。2025年12月，MCP被捐赠给Linux基金会旗下的Agentic AI Foundation，获得了包括OpenAI、Google、Microsoft、AWS等主流AI公司的支持，成为工具集成的事实标准。

MCP的生态系统目前已经相当丰富，社区已经发布了数千个MCP服务器，涵盖GitHub、Google Drive、Slack、数据库、Stripe等各类常用服务，我们可以直接复用许多现成的服务器，而不必为每个工具单独编写集成代码，这也是MCP的意义所在。

安装依赖

MCP支持不在LangChain核心包中，而是需要通过官方维护的langchain-mcp-adapters包来接入MCP工具。

uv add langchain-mcp-adapters

MCP 传输协议

MCP目前支持两种传输协议。

stdio（标准输入输出）：客户端以子进程方式启动MCP服务端程序，通过标准输入输出进行通信，适合本地工具服务器或命令行工具的集成场景。

HTTP（Streamable HTTP）：客户端通过HTTP请求与远程MCP服务端通信，适合Web服务、远程API等场景。注意，旧版本中使用的SSE（Server-Sent Events）作为独立传输协议已在MCP协议版本2025-03-26中弃用，现已被Streamable HTTP取代。

对于MCP使用的传输协议，官方最新版文档明确只列出这两种，但实际上社区早期可能还广泛使用了一些过时或第三方扩展的协议，例如SSE、WebSocket等，langchain-mcp-adapters实际上支持stdio、Streamable HTTP、SSE、WebSocket四种传输协议，不过未来情况可能发生变化，如果用到这些扩展，最好参考LangChain官方文档确认现在是否仍在支持。

创建本地MCP服务器

在演示接入MCP之前，我们得先有一个MCP服务端，这里我们先用fastmcp创建一个简单的本地MCP服务器作为示例。fastmcp是一套基于Python的MCP Server开发框架，有关它的使用不是学习LangChain的重点，我们这里不多介绍，仅简单演示使用。

首先安装fastmcp相关依赖。

uv add fastmcp

然后创建weather_server.py编写以下代码。注意编写完成后，我们不必手动启动它，stdio模式的MCP Server一般都由客户端启动并连接标准输入输出进行通信。

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Math")


@mcp.tool()
def get_weather(city: str) -> str:
    """Get current weather for a given city.

    Args:
        city: City name, for example "Los Angeles" or "Beijing".

    Returns:
        A weather summary string, for example "Sunny 14°C".
    """
    # 实际项目中这里应调用真实的天气API
    return "晴 14℃"


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

通过stdio接入MCP工具

LangChain中，通过stdio接入MCP工具时，我们要添加MCP Server的启动参数，LangChain框架会帮我们启动它。连接MCP Server后，我们可以使用load_mcp_tools()函数从MCP服务器加载工具，并将其自动转换为LangChain兼容的工具对象，此时就可以接入智能体并调用了，下面是例子代码。

import asyncio

from langchain_openai import ChatOpenAI
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp_adapters.tools import load_mcp_tools
from langchain.agents import create_agent

server_params = StdioServerParameters(
    command="C:\\Users\\HUAWEI\\workspace\\workspace-me\\demo-langchain\\.venv\\Scripts\\python.exe",
    args=["-u", "C:\\Users\\HUAWEI\\workspace\\workspace-me\\demo-langchain\\weather_server.py"],
)


async def main():
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 初始化连接
            await session.initialize()

            # 加载MCP工具，自动转换为LangChain工具
            tools = await load_mcp_tools(session)
            print(f"已加载 {len(tools)} 个工具：{[t.name for t in tools]}")

            # 将MCP工具绑定到智能体
            model = ChatOpenAI(
                model="qwen3:30b-a3b",
                base_url="http://localhost:11434/v1/",
                api_key="dummy",
                temperature=1,
                top_p=1,
                max_tokens=16384,
                timeout=120,
                max_retries=6
            )
            agent = create_agent(model, tools)
            resp = await agent.ainvoke({"messages": "洛杉矶现在的天气怎么样？"})
            print(resp["messages"][-1].content)


asyncio.run(main())

注：Python的-u参数用于禁用标准输入输出的缓冲区，以保证流式输出的实时性。

使用MultiServerMCPClient连接多个服务器

实际应用中，如果我们需要同时连接多个MCP服务器，MultiServerMCPClient提供了统一的多服务器管理能力，能让我们同时连接stdio和HTTP类型的服务器，并将所有工具合并为一个列表。

client = MultiServerMCPClient(
    {
        "math": {
            "transport": "stdio",
            "command": "C:\\Users\\HUAWEI\\workspace\\workspace-me\\demo-langchain\\.venv\\Scripts\\python.exe",
            "args": ["-u", "C:\\Users\\HUAWEI\\workspace\\workspace-me\\demo-langchain\\weather_server.py"],
        },
        "weather": {
            "transport": "http",
            "url": "http://localhost:8000/mcp",
            "headers": {
                "Authorization": "Bearer <YOUR_TOKEN>",
            },
        },
    }
)
tools = await client.get_tools()