工具集成:让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 时,整个流程如下:
- 意图识别:Agent 分析用户消息,判断是否需要调用工具。如果用户问"北京今天天气怎么样",Agent 会识别出需要调用天气查询工具。
- 参数构造:Agent 根据用户消息和工具参数定义,自动构造工具调用参数。例如,从"北京今天天气怎么样"中提取出
city="北京"。 - 参数验证:OpenClaw 框架验证参数是否符合工具的模式定义。如果验证失败,Agent 会收到错误信息并尝试修正。
- 工具执行:参数验证通过后,框架调用工具的
execute方法,执行实际的业务逻辑(如发送 HTTP 请求)。 - 结果处理:工具返回结果后,Agent 将结果整合到回复中,生成最终的用户响应。
- 循环调用:如果任务需要多个工具协同完成,Agent 可能会连续调用多个工具,直到任务完成。框架通过
max_tool_calls限制最大调用次数,防止无限循环。
测试工具集成
完成工具集成后,使用以下测试用例验证 Agent 的工具调用行为是否正确。测试"北京今天天气怎么样"来验证天气工具,测试"搜索最新的AI新闻"来验证搜索工具,测试"查询最近10笔订单"来验证数据库工具。同时,也要测试边界情况,比如输入不存在的城市名、发送恶意的 SQL 语句等,确保工具的安全防护机制正常工作。
工具集成是让 Agent 从"纸上谈兵"到"实战应用"的关键一步。掌握了工具集成技术,你的 Agent 就能与真实世界的数据和服务进行交互,成为一个真正有用的智能助手。在下一节教程中,我们将学习如何部署多个 Agent 协同工作,构建更强大的智能体团队。