OpenClaw深度解析（二）：技能开发

本文适合：想扩展 OpenClaw 能力的开发者 前置知识：Python 基础、OpenClaw 架构基础

目录

1. Skill 系统概述

1.1 什么是 Skill？

Skill 是 OpenClaw 的能力扩展包，让 Agent 学会新技能。

类比：

• Chrome 扩展 → 扩展浏览器功能
• VSCode 插件 → 扩展编辑器功能
• OpenClaw Skill → 扩展 Agent 能力

1.2 Skill 能做什么？

1.3 Skill 目录结构

my-skill/
├── SKILL.md # 必需：技能说明文档
├── tools/ # 工具函数目录
│ ├── __init__.py
│ ├── query.py # 查询功能
│ └── update.py # 更新功能
├── scripts/ # 辅助脚本
│ └── setup.py
├── references/ # 参考资料
│ └── api_doc.md
└── tests/ # 测试用例
└── test_query.py

2. 开发第一个 Skill

2.1 案例：天气查询 Skill

目标：让 Agent 能查询任意城市天气

2.2 创建目录

cd ~/.openclaw/workspace/skills
mkdir weather-skill
cd weather-skill
mkdir tools scripts references tests

2.3 编写 SKILL.md

# Weather Skill – 天气查询

## 功能描述
查询全球任意城市的实时天气和天气预报

## 触发条件
用户提到以下关键词时自动触发：
– "天气"
– "气温"
– "下雨"
– "预报"

## 可用工具

### weather_query(city, date="today")
查询指定城市天气

**参数**：
– `city` (str): 城市名，如"北京"、"New York"
– `date` (str): 日期，"today" 或 "YYYY-MM-DD"

**返回**：
```json
{
"city": "北京",
"date": "2026-03-16",
"condition": "晴",
"temperature": {"min": 5, "max": 18},
"humidity": 45,
"wind": {"direction": "西北", "speed": 12}
}

示例：

weather_query("北京")
weather_query("上海", "2026-03-17")

数据源

• 国内城市：中国天气网
• 国际城市：Open-Meteo API

配置项

{
"default_units": "metric", // metric/imperial
"language": "zh-CN"
}

### 2.4 实现工具函数

**tools/query.py**：
```python
"""
天气查询工具函数
"""

import requests
from typing import Dict, Optional

class WeatherAPI:
"""天气 API 封装"""

def __init__(self):
self.base_url = "https://api.open-meteo.com/v1/forecast"

def get_coordinates(self, city: str) -> tuple:
"""
获取城市经纬度
实际项目中可调用地理编码 API
这里用简化版本
"""
# 简化示例，实际应调用地理编码服务
coordinates = {
"北京": (39.9042, 116.4074),
"上海": (31.2304, 121.4737),
"广州": (23.1291, 113.2644),
"深圳": (22.5431, 114.0579),
}
return coordinates.get(city, (39.9042, 116.4074)) # 默认北京

def query(self, city: str, date: str = "today") -> Dict:
"""
查询天气

Args:
city: 城市名
date: 日期（today 或具体日期）

Returns:
天气数据字典
"""
lat, lon = self.get_coordinates(city)

params = {
"latitude": lat,
"longitude": lon,
"current_weather": True,
"daily": ["temperature_2m_max", "temperature_2m_min", "precipitation_sum"],
"timezone": "Asia/Shanghai"
}

try:
response = requests.get(self.base_url, params=params, timeout=10)
response.raise_for_status()
data = response.json()

# 解析数据
current = data.get("current_weather", {})
daily = data.get("daily", {})

result = {
"city": city,
"date": date,
"condition": self._parse_weather_code(current.get("weathercode", 0)),
"temperature": {
"current": current.get("temperature", 0),
"max": daily.get("temperature_2m_max", [0])[0],
"min": daily.get("temperature_2m_min", [0])[0]
},
"wind": {
"direction": current.get("winddirection", 0),
"speed": current.get("windspeed", 0)
}
}

return result

except Exception as e:
return {
"error": str(e),
"city": city
}

def _parse_weather_code(self, code: int) -> str:
"""解析天气代码"""
weather_map = {
0: "晴",
1: "多云",
2: "阴",
3: "雾",
45: "雾",
48: "雾凇",
51: "毛毛雨",
53: "小雨",
55: "大雨",
61: "小雨",
63: "中雨",
65: "大雨",
71: "小雪",
73: "中雪",
75: "大雪",
95: "雷阵雨"
}
return weather_map.get(code, "未知")

# 导出函数供 Agent 调用
def weather_query(city: str, date: str = "today") -> Dict:
"""
天气查询函数（Agent 调用入口）

Args:
city: 城市名
date: 日期

Returns:
天气数据
"""
api = WeatherAPI()
return api.query(city, date)

2.5 注册 Skill

在 openclaw.json 中添加：

{
"skills": {
"enabled": ["weather-skill"],
"paths": {
"weather-skill": "~/.openclaw/workspace/skills/weather-skill"
}
}
}

2.6 测试 Skill

# 重启 Gateway
openclaw gateway restart

# 测试对话
openclaw chat "北京今天天气怎么样？"

预期输出：

北京今天天气：晴，气温 5-18°C，西北风 12km/h。

3. Skill 规范详解

3.1 SKILL.md 必选字段

# [Skill 名称]

## 功能描述
[清晰描述 Skill 的功能和用途]

## 触发条件
[列出触发关键词或场景]

## 可用工具
[列出所有可被 Agent 调用的函数]

### [函数名](参数列表)
[函数说明]

**参数**：
– `param1` (类型): 说明

**返回**：
[返回数据结构]

**示例**：
```python
[调用示例代码]

配置项

[可选的配置参数]

依赖

[外部依赖，如 API Key、库等]

注意事项

[使用限制、警告等]

### 3.2 工具函数命名规范

**推荐格式**：`[领域]_[功能]_[动作]`

| 领域 | 功能 | 动作 | 示例 |
|——|——|——|——|
| `weather` | `query` | – | `weather_query` |
| `stock` | `price` | `get` | `stock_price_get` |
| `file` | `pdf` | `parse` | `file_pdf_parse` |
| `email` | `send` | – | `email_send` |

### 3.3 错误处理规范

```python
def safe_function(param: str) -> Dict:
"""
安全的函数实现

Returns:
成功：{"success": True, "data": …}
失败：{"success": False, "error": "错误信息"}
"""
try:
# 业务逻辑
result = do_something(param)
return {"success": True, "data": result}
except Exception as e:
return {"success": False, "error": str(e)}

4. 工具函数开发实战

4.1 案例：股票数据查询

需求：查询 A 股股票实时行情

tools/stock_query.py：

"""
股票数据查询工具
"""

import requests
from typing import Dict, List
from datetime import datetime

class StockAPI:
"""股票 API 封装"""

def __init__(self):
# 使用新浪财经 API（免费）
self.base_url = "http://hq.sinajs.cn/list="

def get_stock_info(self, symbol: str) –> Dict:
"""
获取股票实时行情

Args:
symbol: 股票代码，如"600519"

Returns:
股票数据
"""
# A 股代码前缀
if symbol.startswith("6"):
code = f"sh{symbol}"
else:
code = f"sz{symbol}"

try:
response = requests.get(
self.base_url + code,
timeout=5,
headers={"User-Agent": "Mozilla/5.0"}
)
response.encoding = "gbk" # 新浪返回 GBK 编码

# 解析返回数据
# 格式：var hq_str_sh600519="贵州茅台，1234.56,…"
content = response.text.strip()
if "=" in content:
data_str = content.split("=")[1].strip('"')
fields = data_str.split(",")

if len(fields) >= 32:
return {
"symbol": symbol,
"name": fields[0],
"current": float(fields[3]),
"open": float(fields[1]),
"high": float(fields[4]),
"low": float[5],
"volume": float(fields[8]),
"amount": float(fields[9]),
"time": f"{fields[30]} {fields[31]}",
"change_percent": round(
(float(fields[3]) – float(fields[2])) / float(fields[2]) * 100, 2
)
}

return {"error": "数据解析失败"}

except Exception as e:
return {"error": str(e)}

def get_stock_history(
self,
symbol: str,
start_date: str,
end_date: str,
period: str = "daily"
) –> List[Dict]:
"""
获取历史行情

Args:
symbol: 股票代码
start_date: 开始日期 YYYY-MM-DD
end_date: 结束日期 YYYY-MM-DD
period: 周期 (daily/weekly/monthly)

Returns:
历史数据列表
"""
# 实现略（可调用聚宽、Tushare 等 API）
pass

# Agent 调用入口
def stock_query(symbol: str) –> Dict:
"""查询股票实时行情"""
api = StockAPI()
return api.get_stock_info(symbol)

def stock_history(
symbol: str,
start_date: str,
end_date: str
) –> List[Dict]:
"""查询股票历史行情"""
api = StockAPI()
return api.get_stock_history(symbol, start_date, end_date)

4.2 案例：文件批量处理

tools/file_batch.py：

"""
文件批量处理工具
"""

import os
import shutil
from pathlib import Path
from typing import List, Dict
from datetime import datetime

class FileBatchProcessor:
"""文件批量处理器"""

def __init__(self, workspace: str):
self.workspace = Path(workspace)

def find_by_extension(self, ext: str) –> List[Path]:
"""
按扩展名查找文件

Args:
ext: 扩展名（不含点），如"pdf"

Returns:
文件路径列表
"""
files = []
for path in self.workspace.rglob(f"*.{ext}"):
if path.is_file():
files.append(path)
return files

def move_to_folder(
self,
files: List[Path],
target_folder: str
) –> Dict:
"""
批量移动文件到文件夹

Args:
files: 文件列表
target_folder: 目标文件夹名

Returns:
操作结果
"""
target = self.workspace / target_folder
target.mkdir(exist_ok=True)

moved = []
errors = []

for file in files:
try:
shutil.move(str(file), str(target / file.name))
moved.append(file.name)
except Exception as e:
errors.append({"file": file.name, "error": str(e)})

return {
"success": len(errors) == 0,
"moved": moved,
"errors": errors,
"target": str(target)
}

def backup_files(
self,
pattern: str,
backup_suffix: str = ".bak"
) –> Dict:
"""
批量备份文件

Args:
pattern: 文件匹配模式，如"*.py"
backup_suffix: 备份后缀

Returns:
备份结果
"""
files = list(self.workspace.glob(pattern))
backed_up = []

for file in files:
backup_path = file.with_suffix(file.suffix + backup_suffix)
shutil.copy2(file, backup_path)
backed_up.append(file.name)

return {
"success": True,
"backed_up": backed_up,
"count": len(backed_up)
}

# Agent 调用入口
def file_find(ext: str) –> List[str]:
"""查找指定扩展名的文件"""
processor = FileBatchProcessor(os.environ.get("WORKSPACE", "."))
files = processor.find_by_extension(ext)
return [str(f) for f in files]

def file_backup(pattern: str) –> Dict:
"""批量备份文件"""
processor = FileBatchProcessor(os.environ.get("WORKSPACE", "."))
return processor.backup_files(pattern)

5. 集成外部 API

5.1 API Key 管理

安全做法：

import os
from pathlib import Path

def get_api_key(service: str) –> str:
"""
安全获取 API Key

优先级：
1. 环境变量
2. 配置文件
3. 抛出异常
"""
# 方法 1：环境变量
key = os.environ.get(f"{service.upper()}_API_KEY")
if key:
return key

# 方法 2：配置文件
config_path = Path.home() / ".openclaw" / "api_keys.json"
if config_path.exists():
import json
with open(config_path) as f:
config = json.load(f)
return config.get(service)

raise ValueError(f"{service} API Key not found")

5.2 调用 REST API

import requests
from typing import Dict, Optional

class APIClient:
"""通用 API 客户端"""

def __init__(self, base_url: str, api_key: str):
self.base_url = base_url
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})

def get(self, endpoint: str, params: Optional[Dict] = None) –> Dict:
"""GET 请求"""
url = f"{self.base_url}/{endpoint}"
response = self.session.get(url, params=params, timeout=30)
response.raise_for_status()
return response.json()

def post(self, endpoint: str, data: Dict) –> Dict:
"""POST 请求"""
url = f"{self.base_url}/{endpoint}"
response = self.session.post(url, json=data, timeout=30)
response.raise_for_status()
return response.json()

def upload_file(self, endpoint: str, file_path: str) –> Dict:
"""文件上传"""
url = f"{self.base_url}/{endpoint}"
with open(file_path, "rb") as f:
files = {"file": f}
response = self.session.post(url, files=files, timeout=60)
response.raise_for_status()
return response.json()

5.3 处理 API 限流

import time
from functools import wraps

def rate_limit(calls: int, period: int):
"""
API 限流装饰器

Args:
calls: 允许调用次数
period: 时间窗口（秒）
"""
timestamps = []

def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
nonlocal timestamps

now = time.time()
# 移除过期时间戳
timestamps = [t for t in timestamps if now – t < period]

if len(timestamps) >= calls:
wait_time = period – (now – timestamps[0])
time.sleep(wait_time)

timestamps.append(time.time())
return func(*args, **kwargs)

return wrapper
return decorator

# 使用示例
@rate_limit(calls=10, period=60) # 每分钟最多 10 次
def api_call():
pass

6. 调试与测试

6.1 本地测试工具函数

tests/test_weather.py：

"""
天气工具测试
"""

import unittest
from tools.query import weather_query

class TestWeatherQuery(unittest.TestCase):

def test_beijing(self):
"""测试北京天气查询"""
result = weather_query("北京")
self.assertIn("city", result)
self.assertEqual(result["city"], "北京")
self.assertIn("temperature", result)

def test_invalid_city(self):
"""测试无效城市"""
result = weather_query("InvalidCity123")
# 应该返回默认数据或错误
self.assertIsInstance(result, dict)

def test_response_structure(self):
"""测试返回数据结构"""
result = weather_query("上海")
required_keys = ["city", "date", "condition", "temperature"]
for key in required_keys:
self.assertIn(key, result)

if __name__ == "__main__":
unittest.main()

6.2 运行测试

# 运行单个测试文件
python tests/test_weather.py

# 运行所有测试
python -m pytest tests/

# 查看覆盖率
python -m pytest tests/ –cov=tools

6.3 调试技巧

添加日志：

import logging

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

def weather_query(city: str):
logger.debug(f"查询天气：{city}")
# …
logger.info(f"查询成功：{result}")

使用断点：

def debug_function():
import pdb; pdb.set_trace() # 断点
# … 代码

7. 发布与分享

7.1 打包 Skill

# 创建压缩包
cd ~/.openclaw/workspace/skills/weather-skill
tar -czf weather-skill.tar.gz \\
SKILL.md \\
tools/ \\
scripts/

# 或创建 git 仓库
git init
git add .
git commit -m "Initial commit"
git remote add origin https://github.com/yourname/weather-skill.git
git push -u origin main

7.2 发布到 ClawHub

ClawHub 是 OpenClaw 的官方技能市场。

步骤：

7.3 分享文档

README.md 模板：

# Weather Skill

## 快速开始

```bash
# 安装
git clone https://github.com/yourname/weather-skill.git
cp -r weather-skill ~/.openclaw/workspace/skills/

# 配置
# 在 openclaw.json 中添加技能路径

# 使用
openclaw chat "北京天气怎么样？"

功能

• 实时天气查询
• 3 天天气预报
• 空气质量指数

配置

{
"api_key": "your-key-here",
"units": "metric"
}

总结

开发 Skill 的核心步骤：

作者：阿财 | 更新时间：2026-03-16