工具集成:让Agent调用外部API

一个没有工具的 AI Agent 就像一个只会思考却无法行动的大脑——它能理解问题,但无法与真实世界交互。OpenClaw 的工具集成系统让 Agent 能够调用外部 API、查询数据库、操作文件系统,真正将 AI 的推理能力转化为实际行动。本教程将手把手教你为 Agent 注册和集成各种外部工具,从天气查询到网页搜索再到数据库操作。

工具集成的核心思想

在 OpenClaw 中,工具(Tool)是 Agent 与外部世界交互的桥梁。每个工具都是一个独立的函数,具有清晰的输入参数定义和输出格式描述。Agent 通过大语言模型的函数调用(Function Calling)能力,自动判断何时需要调用哪个工具,并构造合适的参数。整个过程对用户来说是透明的。

工具注册基础

在 OpenClaw 中注册一个工具需要三个核心要素:工具的描述信息、参数模式定义和执行函数。描述信息告诉 Agent 这个工具能做什么,参数模式定义了工具需要什么输入,执行函数则包含实际的业务逻辑。

# tools/weather_tool.py
from openclaw import Tool, ToolParameter

class WeatherTool(Tool):
    """天气查询工具 —— 获取指定城市的实时天气信息"""

    name = "get_weather"
    description = "查询指定城市的当前天气情况,包括温度、湿度、风速和天气状况。当用户询问天气相关问题时使用此工具。"

    # 定义工具的参数模式
    parameters = [
        ToolParameter(
            name="city",
            type="string",
            description="要查询天气的城市名称,例如:北京、上海、广州",
            required=True
        ),
        ToolParameter(
            name="unit",
            type="string",
            description="温度单位:celsius(摄氏度)或 fahrenheit(华氏度)",
            default="celsius",
            required=False
        )
    ]

    async def execute(self, city: str, unit: str = "celsius") -> dict:
        """执行天气查询"""
        import aiohttp

        api_key = self.config.get("weather_api_key")
        url = f"https://api.weatherapi.com/v1/current.json"
        params = {"key": api_key, "q": city, "aqi": "no"}

        async with aiohttp.ClientSession() as session:
            async with session.get(url, params=params) as resp:
                data = await resp.json()

                if resp.status != 200:
                    return {"error": f"查询失败:{data.get('error', {}).get('message', '未知错误')}"}

                current = data["current"]
                temp = current["temp_c"] if unit == "celsius" else current["temp_f"]
                unit_symbol = "°C" if unit == "celsius" else "°F"

                return {
                    "city": city,
                    "temperature": f"{temp}{unit_symbol}",
                    "humidity": f"{current['humidity']}%",
                    "wind_speed": f"{current['wind_kph']} km/h",
                    "condition": current["condition"]["text"],
                    "last_updated": current["last_updated"]
                }

上面的代码定义了一个完整的天气查询工具。注意 description 字段的写法——它不仅是给人看的文档,更是 Agent 理解工具用途的关键依据。一个好的工具描述应该清楚地说明:这个工具能做什么、什么时候应该使用它、输入和输出是什么。

参数模式与数据验证

OpenClaw 的工具系统内置了强大的参数验证机制。当 Agent 决定调用一个工具时,框架会自动验证传入的参数是否符合预定义的模式。如果参数缺失或类型不匹配,框架会返回清晰的错误信息给 Agent,让 Agent 有机会修正参数后重新调用。

参数验证支持以下类型:string(字符串)、integer(整数)、float(浮点数)、boolean(布尔值)、array(数组)和 object(对象)。对于字符串类型的参数,你还可以指定 enum 来限制可选值范围。

安全提示

永远不要信任来自 Agent 的工具参数。即使参数通过了类型验证,你仍然需要在 execute 方法中进行业务逻辑层面的校验。例如,对城市名称进行白名单检查、对 SQL 查询进行注入防护、对文件路径进行越权访问检测等。工具是 Agent 与外部世界的接口,也是安全防护的重要关卡。

实战:集成网页搜索工具

网页搜索是 AI Agent 最常用的工具之一。它让 Agent 能够获取最新的互联网信息,弥补大语言模型知识截止日期的局限性。下面我们来集成一个基于搜索 API 的网页搜索工具。

# tools/web_search.py
from openclaw import Tool, ToolParameter

class WebSearchTool(Tool):
    """网页搜索工具 —— 搜索互联网获取最新信息"""

    name = "web_search"
    description = "在互联网上搜索信息。当用户询问最新事件、实时数据或你的知识可能过时的话题时使用此工具。返回搜索结果的标题、摘要和链接。"

    parameters = [
        ToolParameter(
            name="query",
            type="string",
            description="搜索关键词,建议使用简洁精准的查询语句",
            required=True
        ),
        ToolParameter(
            name="num_results",
            type="integer",
            description="返回的搜索结果数量,默认5条,最多10条",
            default=5,
            required=False
        ),
        ToolParameter(
            name="language",
            type="string",
            description="搜索结果的语言偏好,如 zh-CN, en-US",
            default="zh-CN",
            required=False
        )
    ]

    async def execute(self, query: str, num_results: int = 5,
                      language: str = "zh-CN") -> dict:
        """执行网页搜索"""
        import aiohttp

        api_key = self.config.get("search_api_key")
        url = "https://api.searchapi.io/v1/search"
        params = {
            "q": query,
            "num": min(num_results, 10),
            "hl": language,
            "api_key": api_key
        }

        async with aiohttp.ClientSession() as session:
            async with session.get(url, params=params) as resp:
                data = await resp.json()

                results = []
                for item in data.get("organic_results", [])[:num_results]:
                    results.append({
                        "title": item.get("title", ""),
                        "snippet": item.get("snippet", ""),
                        "url": item.get("link", ""),
                        "source": item.get("source", "")
                    })

                return {
                    "query": query,
                    "total_found": len(results),
                    "results": results
                }

实战:集成数据库查询工具

在企业应用中,Agent 经常需要查询数据库来获取业务数据。下面我们来创建一个安全的数据库查询工具,它支持只读的 SQL 查询,并内置了安全限制。

# tools/database_query.py
from openclaw import Tool, ToolParameter
import asyncpg

class DatabaseQueryTool(Tool):
    """数据库查询工具 —— 安全地查询业务数据库"""

    name = "query_database"
    description = "在业务数据库中执行只读SQL查询。适用于查询订单信息、用户数据、统计数据等。仅支持SELECT语句,禁止修改数据。"

    parameters = [
        ToolParameter(
            name="sql",
            type="string",
            description="要执行的SQL查询语句,仅支持SELECT语句",
            required=True
        ),
        ToolParameter(
            name="max_rows",
            type="integer",
            description="返回的最大行数,默认100",
            default=100,
            required=False
        )
    ]

    # 安全白名单:只允许查询这些表
    ALLOWED_TABLES = {"orders", "products", "users", "statistics"}
    # 禁止的SQL关键字
    FORBIDDEN_KEYWORDS = {"INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "CREATE", "TRUNCATE"}

    async def execute(self, sql: str, max_rows: int = 100) -> dict:
        """执行数据库查询"""
        # 安全检查:确保是只读查询
        sql_upper = sql.upper().strip()
        if not sql_upper.startswith("SELECT"):
            return {"error": "安全限制:仅允许 SELECT 查询语句"}

        for keyword in self.FORBIDDEN_KEYWORDS:
            if keyword in sql_upper:
                return {"error": f"安全限制:检测到禁止的SQL关键字 '{keyword}'"}

        # 执行查询
        try:
            conn = await asyncpg.connect(
                host=self.config["db_host"],
                port=self.config["db_port"],
                database=self.config["db_name"],
                user=self.config["db_user"],
                password=self.config["db_password"]
            )

            # 使用 LIMIT 限制返回行数
            safe_sql = f"SELECT * FROM ({sql}) AS subquery LIMIT {max_rows}"
            rows = await conn.fetch(safe_sql)
            await conn.close()

            # 将结果转换为字典列表
            results = [dict(row) for row in rows]

            return {
                "row_count": len(results),
                "columns": list(results[0].keys()) if results else [],
                "data": results
            }

        except Exception as e:
            return {"error": f"查询执行失败:{str(e)}"}

将工具绑定到 Agent

工具定义完成后,需要将它们注册到 Agent 上。在 OpenClaw 中,你可以通过配置文件或代码两种方式来完成工具绑定。推荐使用配置文件的方式,因为它更灵活,可以在不修改代码的情况下调整 Agent 的工具集。

# config.yaml - 将工具绑定到 Agent
agents:
  - name: "智能助手"
    module: "agents.assistant"
    class: "MyFirstAgent"
    tools:
      - name: "get_weather"
        module: "tools.weather_tool"
        class: "WeatherTool"
        config:
          weather_api_key: "${WEATHER_API_KEY}"

      - name: "web_search"
        module: "tools.web_search"
        class: "WebSearchTool"
        config:
          search_api_key: "${SEARCH_API_KEY}"

      - name: "query_database"
        module: "tools.database_query"
        class: "DatabaseQueryTool"
        config:
          db_host: "localhost"
          db_port: 5432
          db_name: "myapp"
          db_user: "${DB_USER}"
          db_password: "${DB_PASSWORD}"

    settings:
      model: "gpt-4"
      temperature: 0.7
      # 启用自动工具调用
      auto_tool_call: true
      # 最大连续工具调用次数(防止无限循环)
      max_tool_calls: 10

工具描述的最佳实践

工具描述的质量直接影响 Agent 调用工具的准确性。编写工具描述时,请遵循以下原则:第一,明确说明工具的功能和使用场景;第二,描述每个参数的含义和约束条件;第三,如果工具有副作用(如写入数据),一定要在描述中说明;第四,提供参数值的示例,帮助 Agent 构造正确的调用参数。好的工具描述就像好的函数文档——让调用者一看就知道该怎么用。

工具调用的执行流程

理解工具调用的完整执行流程,有助于你更好地调试和优化 Agent 的工具使用行为。当用户发送一条消息给 Agent 时,整个流程如下:

  1. 意图识别:Agent 分析用户消息,判断是否需要调用工具。如果用户问"北京今天天气怎么样",Agent 会识别出需要调用天气查询工具。
  2. 参数构造:Agent 根据用户消息和工具参数定义,自动构造工具调用参数。例如,从"北京今天天气怎么样"中提取出 city="北京"。
  3. 参数验证:OpenClaw 框架验证参数是否符合工具的模式定义。如果验证失败,Agent 会收到错误信息并尝试修正。
  4. 工具执行:参数验证通过后,框架调用工具的 execute 方法,执行实际的业务逻辑(如发送 HTTP 请求)。
  5. 结果处理:工具返回结果后,Agent 将结果整合到回复中,生成最终的用户响应。
  6. 循环调用:如果任务需要多个工具协同完成,Agent 可能会连续调用多个工具,直到任务完成。框架通过 max_tool_calls 限制最大调用次数,防止无限循环。

测试工具集成

完成工具集成后,使用以下测试用例验证 Agent 的工具调用行为是否正确。测试"北京今天天气怎么样"来验证天气工具,测试"搜索最新的AI新闻"来验证搜索工具,测试"查询最近10笔订单"来验证数据库工具。同时,也要测试边界情况,比如输入不存在的城市名、发送恶意的 SQL 语句等,确保工具的安全防护机制正常工作。

工具集成是让 Agent 从"纸上谈兵"到"实战应用"的关键一步。掌握了工具集成技术,你的 Agent 就能与真实世界的数据和服务进行交互,成为一个真正有用的智能助手。在下一节教程中,我们将学习如何部署多个 Agent 协同工作,构建更强大的智能体团队。