夜雨聆风本文适合：想扩展 OpenClaw 能力的开发者前置知识：Python 基础、OpenClaw 架构基础
目录
1. Skill 系统概述 2. 开发第一个 Skill 3. Skill 规范详解 4. 工具函数开发实战 5. 集成外部 API 6. 调试与测试 7. 发布与分享
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.py2. 开发第一个 Skill
2.1 案例：天气查询 Skill
目标：让 Agent 能查询任意城市天气
2.2 创建目录
cd ~/.openclaw/workspace/skillsmkdir weather-skillcd weather-skillmkdir tools scripts references tests2.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 requestsfrom typing import Dict, Optionalclass 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
# 重启 Gatewayopenclaw 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 错误处理规范```pythondef 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 requestsfrom typing importDict, Listfrom datetime import datetimeclassStockAPI:"""股票 API 封装"""def__init__(self):# 使用新浪财经 API（免费）self.base_url = "http://hq.sinajs.cn/list="defget_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(",")iflen(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)}defget_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 调用入口defstock_query(symbol: str) -> Dict:"""查询股票实时行情""" api = StockAPI()return api.get_stock_info(symbol)defstock_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 osimport shutilfrom pathlib import Pathfrom typing importList, Dictfrom datetime import datetimeclassFileBatchProcessor:"""文件批量处理器"""def__init__(self, workspace: str):self.workspace = Path(workspace)deffind_by_extension(self, ext: str) -> List[Path]:""" 按扩展名查找文件 Args: ext: 扩展名（不含点），如"pdf" Returns: 文件路径列表 """ files = []for path inself.workspace.rglob(f"*.{ext}"):if path.is_file(): files.append(path)return filesdefmove_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) }defbackup_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 调用入口deffile_find(ext: str) -> List[str]:"""查找指定扩展名的文件""" processor = FileBatchProcessor(os.environ.get("WORKSPACE", ".")) files = processor.find_by_extension(ext)return [str(f) for f in files]deffile_backup(pattern: str) -> Dict:"""批量备份文件""" processor = FileBatchProcessor(os.environ.get("WORKSPACE", "."))return processor.backup_files(pattern)5. 集成外部 API
5.1 API Key 管理
安全做法：
import osfrom pathlib import Pathdefget_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 jsonwithopen(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 requestsfrom typing importDict, OptionalclassAPIClient:"""通用 API 客户端"""def__init__(self, base_url: str, api_key: str):self.base_url = base_urlself.session = requests.Session()self.session.headers.update({"Authorization": f"Bearer {api_key}","Content-Type": "application/json" })defget(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()defpost(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()defupload_file(self, endpoint: str, file_path: str) -> Dict:"""文件上传""" url = f"{self.base_url}/{endpoint}"withopen(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 timefrom functools import wrapsdefrate_limit(calls: int, period: int):""" API 限流装饰器 Args: calls: 允许调用次数 period: 时间窗口（秒） """ timestamps = []defdecorator(func): @wraps(func)defwrapper(*args, **kwargs):nonlocal timestamps now = time.time()# 移除过期时间戳 timestamps = [t for t in timestamps if now - t < period]iflen(timestamps) >= calls: wait_time = period - (now - timestamps[0]) time.sleep(wait_time) timestamps.append(time.time())return func(*args, **kwargs)return wrapperreturn decorator# 使用示例@rate_limit(calls=10, period=60) # 每分钟最多 10 次defapi_call():pass6. 调试与测试
6.1 本地测试工具函数
tests/test_weather.py：
"""天气工具测试"""import unittestfrom tools.query import weather_queryclassTestWeatherQuery(unittest.TestCase):deftest_beijing(self):"""测试北京天气查询""" result = weather_query("北京")self.assertIn("city", result)self.assertEqual(result["city"], "北京")self.assertIn("temperature", result)deftest_invalid_city(self):"""测试无效城市""" result = weather_query("InvalidCity123")# 应该返回默认数据或错误self.assertIsInstance(result, dict)deftest_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=tools6.3 调试技巧
添加日志：
import logginglogging.basicConfig(level=logging.DEBUG)logger = logging.getLogger(__name__)defweather_query(city: str): logger.debug(f"查询天气：{city}")# ... logger.info(f"查询成功：{result}")使用断点：
defdebug_function():import pdb; pdb.set_trace() # 断点# ... 代码7. 发布与分享
7.1 打包 Skill
# 创建压缩包cd ~/.openclaw/workspace/skills/weather-skilltar -czf weather-skill.tar.gz \ SKILL.md \ tools/ \ scripts/# 或创建 git 仓库git initgit add .git commit -m "Initial commit"git remote add origin https://github.com/yourname/weather-skill.gitgit push -u origin main7.2 发布到 ClawHub
ClawHub 是 OpenClaw 的官方技能市场。
步骤：
1. 在 clawhub.com 注册账号 2. 创建新 Skill 页面 3. 上传 SKILL.md 和源代码 4. 填写描述、截图、使用文档 5. 提交审核
7.3 分享文档
README.md 模板：
# Weather Skill## 快速开始```bash# 安装git clone https://github.com/yourname/weather-skill.gitcp -r weather-skill ~/.openclaw/workspace/skills/# 配置# 在 openclaw.json 中添加技能路径# 使用openclaw chat "北京天气怎么样？"功能
• 实时天气查询 • 3 天天气预报 • 空气质量指数
配置
{"api_key":"your-key-here","units":"metric"}总结
开发 Skill 的核心步骤：
1. 编写 SKILL.md 文档 2. 实现工具函数 3. 本地测试 4. 注册到 OpenClaw 5. 发布分享
作者：阿财 | 更新时间：2026-03-16