OpenClaw 技能开发入门：1小时教你写出第一个自定义 Skill
这个文件告诉 Agent：这个 Skill 能做什么、接受什么参数、返回什么结果。写好 schema，Agent 才能正确路由和调用。"description": "查询指定城市的实时天气信息，返回温度、天气状况、风力等数据","city": {"description": "城市名称，支持中文（如：北京、上海）或英文（Beijing）",},"unit": {"description": "温度
·
作者：bugyuan
标签：OpenClawSkill开发MCPAI Agent插件系统Python工具扩展
阅读时长：约 22 分钟
前置阅读：本文是 OpenClaw 系列第 5 篇，建议先读《办公自动化场景实战》
代码环境：Python 3.11 · OpenClaw 1.x
你将在这 1 小时里做出什么
一个叫 WeatherBrief 的自定义 Skill：
- 接收城市名，查询实时天气
- 自动生成一段自然语言天气播报（「今天北京晴，气温 12~24℃，出门记得带外套」）
- 在晨间工作流里，把天气播报自动插进每日简报
整个流程走完，你就理解了 OpenClaw Skill 的全部核心机制。
先搞清楚：Skill 是什么，不是什么
回顾一下 OpenClaw 的结构
在前几篇文章里，我们构建了这样一个体系：
用户指令（自然语言）
↓
OpenClaw Agent
↓
Tool Router（意图识别 → 选工具）
↓
MCPTool（具体执行：读邮件、建日程、写报表……）
Skill 是 MCPTool 的升级版。
普通 MCPTool 只是一个函数集合——你定义几个方法，Agent 调用它们。Skill 在此基础上增加了三样东西：
Skill = MCPTool
+ 独立的配置系统（每个 Skill 有自己的 config）
+ 生命周期钩子（初始化、健康检查、清理）
+ Skill 间的依赖声明（A Skill 可以调用 B Skill）
用一句话说：MCPTool 是函数，Skill 是插件。
Skill 的文件结构
my_skill/
├── __init__.py # Skill 入口，声明元信息
├── skill.py # 核心逻辑（继承 BaseSkill）
├── config.yaml # 默认配置（用户可覆盖）
├── schema.json # 入参/出参 JSON Schema（供 Agent 理解）
└── tests/
└── test_skill.py # 单元测试
这是 OpenClaw 约定的结构，遵守它，你的 Skill 才能被框架自动发现和加载。
第一步：理解 BaseSkill 的骨架
在写 WeatherBrief 之前，先把框架给你的"毛坯房"看清楚。
# openclaw/core/base_skill.py（框架源码，理解用）
from abc import ABC, abstractmethod
from typing import Any
import yaml
from pathlib import Path
class BaseSkill(ABC):
"""
所有自定义 Skill 的基类。
继承它，实现抽象方法，你的 Skill 就能接入 OpenClaw。
"""
# ── 必填：Skill 基本信息 ──
name: str = "" # Skill 唯一标识（snake_case）
display_name: str = "" # 对用户展示的名称
description: str = "" # 告诉 Agent 这个 Skill 做什么（影响工具路由！）
version: str = "0.1.0"
author: str = ""
# ── 可选：依赖声明 ──
requires: list[str] = [] # 依赖的其他 Skill 名称列表
def __init__(self):
self._config = self._load_config()
self._initialized = False
# ── 生命周期：初始化 ──
def setup(self) -> None:
"""
Skill 首次加载时调用。
适合做：API 连通性测试、数据库连接、缓存预热。
"""
pass # 默认空实现，子类按需覆盖
# ── 生命周期：健康检查 ──
def health_check(self) -> dict:
"""
框架定期调用（每60秒），检查 Skill 是否正常。
返回 {"status": "ok"} 或 {"status": "error", "reason": "..."}
"""
return {"status": "ok"}
# ── 生命周期：清理 ──
def teardown(self) -> None:
"""
Skill 卸载时调用。
适合做：关闭连接、释放资源。
"""
pass
# ── 必须实现：执行入口 ──
@abstractmethod
def execute(self, action: str, params: dict) -> Any:
"""
Agent 调用 Skill 时的统一入口。
action：具体操作名（对应 schema.json 中定义的 actions）
params：操作参数
"""
...
# ── 工具方法：读取配置 ──
def _load_config(self) -> dict:
config_path = Path(__file__).parent / "config.yaml"
if config_path.exists():
with open(config_path) as f:
return yaml.safe_load(f) or {}
return {}
def get_config(self, key: str, default=None):
return self._config.get(key, default)
记住三件事：
- 继承
BaseSkill - 填写
name/description（description 直接影响 Agent 路由准确度） - 实现
execute方法
第二步：写 WeatherBrief Skill
开始动手。创建目录结构：
mkdir -p skills/weather_brief/tests
touch skills/weather_brief/__init__.py
touch skills/weather_brief/skill.py
touch skills/weather_brief/config.yaml
touch skills/weather_brief/schema.json
touch skills/weather_brief/tests/test_skill.py
2.1 先定义 schema.json（定义 Skill 的能力边界）
这个文件告诉 Agent：这个 Skill 能做什么、接受什么参数、返回什么结果。写好 schema，Agent 才能正确路由和调用。
{
"skill": "weather_brief",
"actions": [
{
"name": "get_weather",
"description": "查询指定城市的实时天气信息，返回温度、天气状况、风力等数据",
"parameters": {
"city": {
"type": "string",
"description": "城市名称，支持中文（如：北京、上海）或英文（Beijing）",
"required": true
},
"unit": {
"type": "string",
"description": "温度单位",
"enum": ["celsius", "fahrenheit"],
"default": "celsius",
"required": false
}
},
"returns": {
"city": "城市名",
"temperature": "当前温度（数字）",
"feels_like": "体感温度",
"condition": "天气状况描述",
"humidity": "湿度百分比",
"wind_speed": "风速（km/h）",
"wind_direction": "风向"
}
},
{
"name": "generate_brief",
"description": "生成自然语言风格的天气播报文字，适合插入每日简报或推送通知",
"parameters": {
"city": {
"type": "string",
"description": "城市名称",
"required": true
},
"style": {
"type": "string",
"description": "播报风格",
"enum": ["formal", "casual", "brief"],
"default": "casual",
"required": false
}
},
"returns": {
"brief": "天气播报文字",
"raw_data": "原始天气数据"
}
}
]
}
2.2 配置文件 config.yaml
# skills/weather_brief/config.yaml
# 天气 API 配置（使用 OpenWeatherMap 免费 API）
api:
provider: openweathermap # 支持：openweathermap / qweather
base_url: https://api.openweathermap.org/data/2.5
api_key: "" # 留空，由用户在 openclaw.secrets.yaml 里配置
timeout_seconds: 10
retry_times: 3
# 默认设置
defaults:
unit: celsius
style: casual
language: zh_CN # 天气描述语言
# 缓存配置（同一城市的天气数据，10分钟内复用）
cache:
enabled: true
ttl_seconds: 600
# 播报文字生成（使用 OpenClaw 内置 LLM）
brief_generation:
use_llm: true # false 则用模板生成，不调用 LLM
max_tokens: 100
2.3 核心实现 skill.py
# skills/weather_brief/skill.py
import requests
import time
from typing import Any
from loguru import logger
from openclaw.core import BaseSkill
class WeatherBriefSkill(BaseSkill):
# ── Skill 元信息 ──
name = "weather_brief"
display_name = "天气播报"
description = """
查询实时天气信息并生成自然语言播报。
适用场景：
- 用户询问某城市今天/明天天气
- 每日简报中自动插入天气信息
- 出行建议（是否需要带伞、穿什么）
"""
version = "1.0.0"
author = "bugyuan"
# 不依赖其他 Skill
requires = []
def __init__(self):
super().__init__()
self._cache: dict = {} # 简单内存缓存
self._session = None # requests Session（复用连接）
# ── 生命周期：初始化 ──
def setup(self) -> None:
"""验证 API Key 是否配置，测试连通性"""
api_key = self._get_api_key()
if not api_key:
raise ValueError(
"WeatherBriefSkill 未配置 API Key。\n"
"请在 openclaw.secrets.yaml 中添加：\n"
" weather_brief:\n"
" api_key: your_openweathermap_key\n"
"（免费注册：https://openweathermap.org/api）"
)
# 建立持久 Session（提升请求性能）
self._session = requests.Session()
self._session.headers.update({"User-Agent": "OpenClaw/1.0"})
# 连通性测试（用北京做探针）
try:
test = self._fetch_weather("Beijing", api_key)
logger.success(f"✅ WeatherBriefSkill 初始化成功，"
f"API 连接正常（测试城市：{test['name']}）")
except Exception as e:
raise RuntimeError(f"天气 API 连接失败：{e}")
self._initialized = True
# ── 生命周期：健康检查 ──
def health_check(self) -> dict:
try:
api_key = self._get_api_key()
if not api_key:
return {"status": "error", "reason": "API Key 未配置"}
self._fetch_weather("Beijing", api_key)
cache_size = len(self._cache)
return {
"status": "ok",
"cache_entries": cache_size,
"initialized": self._initialized
}
except Exception as e:
return {"status": "error", "reason": str(e)}
# ── 生命周期：清理 ──
def teardown(self) -> None:
if self._session:
self._session.close()
self._cache.clear()
logger.info("WeatherBriefSkill 已释放资源")
# ── 核心：统一执行入口 ──
def execute(self, action: str, params: dict) -> Any:
"""
Agent 调用入口。
action 对应 schema.json 中定义的操作名。
"""
action_map = {
"get_weather": self._action_get_weather,
"generate_brief": self._action_generate_brief,
}
handler = action_map.get(action)
if not handler:
raise ValueError(
f"未知操作：{action}。"
f"支持的操作：{list(action_map.keys())}"
)
logger.info(f"🌤 WeatherBriefSkill.{action}({params})")
return handler(params)
# ── Action 实现 ──
def _action_get_weather(self, params: dict) -> dict:
"""获取实时天气数据"""
city = params.get("city", "").strip()
unit = params.get("unit", self.get_config("defaults.unit", "celsius"))
if not city:
raise ValueError("city 参数不能为空")
# 检查缓存
cache_key = f"{city}:{unit}"
if self.get_config("cache.enabled", True):
cached = self._get_cache(cache_key)
if cached:
logger.debug(f" 缓存命中：{city}")
return cached
# 调用天气 API
raw = self._fetch_weather(city, self._get_api_key())
# 标准化返回格式
result = self._normalize_weather(raw, unit)
# 写入缓存
self._set_cache(cache_key, result)
return result
def _action_generate_brief(self, params: dict) -> dict:
"""生成天气播报文字"""
city = params.get("city", "").strip()
style = params.get("style",
self.get_config("defaults.style", "casual"))
if not city:
raise ValueError("city 参数不能为空")
# 先获取天气数据
weather_data = self._action_get_weather({"city": city})
# 生成播报文字
use_llm = self.get_config("brief_generation.use_llm", True)
if use_llm:
brief = self._generate_with_llm(weather_data, style)
else:
brief = self._generate_with_template(weather_data, style)
return {
"brief": brief,
"raw_data": weather_data
}
# ── 私有方法 ──
def _fetch_weather(self, city: str, api_key: str) -> dict:
"""调用 OpenWeatherMap API"""
url = f"{self.get_config('api.base_url')}/weather"
params = {
"q": city,
"appid": api_key,
"lang": self.get_config("defaults.language", "zh_cn"),
"units": "metric" # 始终用 metric，转换在 normalize 层处理
}
timeout = self.get_config("api.timeout_seconds", 10)
retry = self.get_config("api.retry_times", 3)
for attempt in range(retry):
try:
resp = self._session.get(url, params=params,
timeout=timeout)
resp.raise_for_status()
return resp.json()
except requests.exceptions.Timeout:
if attempt == retry - 1:
raise TimeoutError(f"天气 API 超时（{timeout}s），已重试 {retry} 次")
time.sleep(1 * (attempt + 1)) # 指数退避
except requests.exceptions.HTTPError as e:
if e.response.status_code == 404:
raise ValueError(f"找不到城市：{city}，"
f"请检查城市名称是否正确")
elif e.response.status_code == 401:
raise PermissionError("API Key 无效，请检查配置")
raise
def _normalize_weather(self, raw: dict, unit: str) -> dict:
"""把 API 原始数据转换为标准格式"""
temp_c = raw["main"]["temp"]
feels_c = raw["main"]["feels_like"]
if unit == "fahrenheit":
temp = round(temp_c * 9 / 5 + 32, 1)
feels = round(feels_c * 9 / 5 + 32, 1)
unit_symbol = "°F"
else:
temp = round(temp_c, 1)
feels = round(feels_c, 1)
unit_symbol = "°C"
return {
"city": raw.get("name", ""),
"country": raw.get("sys", {}).get("country", ""),
"temperature": temp,
"feels_like": feels,
"unit": unit_symbol,
"condition": raw["weather"][0]["description"],
"condition_code": raw["weather"][0]["id"],
"humidity": raw["main"]["humidity"],
"wind_speed": round(raw["wind"]["speed"] * 3.6, 1), # m/s → km/h
"wind_direction": self._wind_degree_to_direction(
raw["wind"].get("deg", 0)
),
"visibility": raw.get("visibility", 0) // 1000, # m → km
}
def _wind_degree_to_direction(self, degree: float) -> str:
"""风向角度转中文方向"""
directions = ["北", "东北", "东", "东南",
"南", "西南", "西", "西北"]
idx = round(degree / 45) % 8
return directions[idx]
def _generate_with_llm(self, weather: dict, style: str) -> str:
"""
使用 OpenClaw 内置 LLM 生成播报文字。
通过 self.llm（框架注入）调用，无需自己管理 LLM 连接。
"""
style_guide = {
"formal": "正式、专业，适合公务场合",
"casual": "轻松、口语化，像朋友告诉你天气",
"brief": "极简，一句话，不超过20字"
}.get(style, "轻松、口语化")
prompt = f"""
根据以下天气数据，生成一段天气播报。
天气数据：
- 城市：{weather['city']}
- 当前温度：{weather['temperature']}{weather['unit']}
- 体感温度：{weather['feels_like']}{weather['unit']}
- 天气状况：{weather['condition']}
- 湿度：{weather['humidity']}%
- 风速：{weather['wind_speed']} km/h，{weather['wind_direction']}风
播报风格：{style_guide}
要求：
- 自然流畅，像真人在说
- 包含实用提示（要不要带伞、穿什么）
- 不超过60字
- 不要重复"天气播报"这几个字
- 直接输出播报内容，不要任何前缀
"""
# self.llm 由 OpenClaw 框架在加载 Skill 时自动注入
return self.llm.chat(user=prompt).strip()
def _generate_with_template(self, weather: dict,
style: str) -> str:
"""
不使用 LLM 的模板生成（零 Token 成本，质量略低）
"""
city = weather["city"]
temp = weather["temperature"]
unit = weather["unit"]
condition = weather["condition"]
humidity = weather["humidity"]
wind = weather["wind_speed"]
# 穿衣建议
if temp < 5:
clothing = "建议穿厚羽绒服"
elif temp < 15:
clothing = "建议穿外套"
elif temp < 25:
clothing = "短袖或薄外套即可"
else:
clothing = "天气较热，注意防晒"
# 带伞建议
rain_codes = range(200, 622) # 雷暴到雪的天气代码
umbrella = "记得带伞" if weather["condition_code"] in rain_codes else ""
if style == "brief":
return f"{city} {condition} {temp}{unit} {clothing}"
elif style == "formal":
return (f"今日{city}天气：{condition}，气温 {temp}{unit}，"
f"湿度 {humidity}%，{wind} km/h 风速。{clothing}。"
f"{'建议携带雨具。' if umbrella else ''}")
else: # casual
brief = f"今天{city}{condition}，温度 {temp}{unit}，{clothing}"
if umbrella:
brief += f"，{umbrella}"
brief += "～"
return brief
# ── 缓存工具方法 ──
def _get_cache(self, key: str):
if key not in self._cache:
return None
entry = self._cache[key]
ttl = self.get_config("cache.ttl_seconds", 600)
if time.time() - entry["ts"] > ttl:
del self._cache[key]
return None
return entry["data"]
def _set_cache(self, key: str, data: dict):
self._cache[key] = {"data": data, "ts": time.time()}
def _get_api_key(self) -> str:
"""
从 OpenClaw 密钥管理器读取 API Key（不硬编码，参考安全篇）。
框架自动注入 self.secrets，由 SecretManager 提供。
"""
# 优先从 secrets 管理器读取
if hasattr(self, "secrets"):
key = self.secrets.get("weather_brief.api_key")
if key:
return key
# 降级：从配置文件读取（开发环境用）
return self.get_config("api.api_key", "")
2.4 入口文件 __init__.py
# skills/weather_brief/__init__.py
from .skill import WeatherBriefSkill
# OpenClaw 框架通过这里发现 Skill
__skill__ = WeatherBriefSkill
__all__ = ["WeatherBriefSkill"]
第三步：写测试（10分钟，别跳过）
跳过测试的 Skill，迟早在工作流里翻车。
# skills/weather_brief/tests/test_skill.py
import pytest
from unittest.mock import patch, MagicMock
from skills.weather_brief import WeatherBriefSkill
# ── Mock 数据 ──
MOCK_API_RESPONSE = {
"name": "Beijing",
"sys": {"country": "CN"},
"main": {
"temp": 15.3,
"feels_like": 13.8,
"humidity": 45
},
"weather": [{"description": "晴", "id": 800}],
"wind": {"speed": 3.2, "deg": 45},
"visibility": 10000
}
MOCK_WEATHER_NORMALIZED = {
"city": "Beijing",
"country": "CN",
"temperature": 15.3,
"feels_like": 13.8,
"unit": "°C",
"condition": "晴",
"condition_code": 800,
"humidity": 45,
"wind_speed": 11.5,
"wind_direction": "东北",
"visibility": 10
}
@pytest.fixture
def skill():
"""创建不真实调用 API 的 Skill 实例"""
s = WeatherBriefSkill()
# 跳过 setup 中的真实 API 调用
s._initialized = True
s._session = MagicMock()
return s
class TestGetWeather:
@patch.object(WeatherBriefSkill, "_fetch_weather",
return_value=MOCK_API_RESPONSE)
@patch.object(WeatherBriefSkill, "_get_api_key",
return_value="fake-key")
def test_basic_query(self, mock_key, mock_fetch, skill):
"""测试基本天气查询"""
result = skill.execute("get_weather", {"city": "北京"})
assert result["city"] == "Beijing"
assert result["temperature"] == 15.3
assert result["unit"] == "°C"
assert result["condition"] == "晴"
@patch.object(WeatherBriefSkill, "_fetch_weather",
return_value=MOCK_API_RESPONSE)
@patch.object(WeatherBriefSkill, "_get_api_key",
return_value="fake-key")
def test_fahrenheit_conversion(self, mock_key, mock_fetch, skill):
"""测试华氏度转换"""
result = skill.execute("get_weather",
{"city": "北京", "unit": "fahrenheit"})
# 15.3°C ≈ 59.5°F
assert abs(result["temperature"] - 59.5) < 0.5
assert result["unit"] == "°F"
def test_empty_city_raises(self, skill):
"""空城市名应该抛出异常"""
with pytest.raises(ValueError, match="city 参数不能为空"):
skill.execute("get_weather", {"city": ""})
def test_unknown_action_raises(self, skill):
"""未知操作应该抛出异常"""
with pytest.raises(ValueError, match="未知操作"):
skill.execute("fly_to_moon", {})
class TestCache:
@patch.object(WeatherBriefSkill, "_fetch_weather",
return_value=MOCK_API_RESPONSE)
@patch.object(WeatherBriefSkill, "_get_api_key",
return_value="fake-key")
def test_cache_hit(self, mock_key, mock_fetch, skill):
"""同一城市第二次查询应命中缓存，不再调用 API"""
skill.execute("get_weather", {"city": "北京"})
skill.execute("get_weather", {"city": "北京"})
# API 只被调用了一次
assert mock_fetch.call_count == 1
@patch.object(WeatherBriefSkill, "_fetch_weather",
return_value=MOCK_API_RESPONSE)
@patch.object(WeatherBriefSkill, "_get_api_key",
return_value="fake-key")
def test_cache_expiry(self, mock_key, mock_fetch, skill):
"""缓存过期后应重新调用 API"""
import time
skill.execute("get_weather", {"city": "北京"})
# 手动让缓存过期
for key in skill._cache:
skill._cache[key]["ts"] -= 700 # 超过 600s TTL
skill.execute("get_weather", {"city": "北京"})
assert mock_fetch.call_count == 2
class TestBriefGeneration:
def test_template_casual(self, skill):
"""模板生成：casual 风格"""
brief = skill._generate_with_template(MOCK_WEATHER_NORMALIZED,
"casual")
assert "Beijing" in brief or "晴" in brief
assert len(brief) < 100
def test_template_brief(self, skill):
"""模板生成：brief 风格不超过 30 字"""
brief = skill._generate_with_template(MOCK_WEATHER_NORMALIZED,
"brief")
assert len(brief) <= 30
def test_wind_direction(self, skill):
"""风向转换测试"""
assert skill._wind_degree_to_direction(0) == "北"
assert skill._wind_degree_to_direction(90) == "东"
assert skill._wind_degree_to_direction(180) == "南"
assert skill._wind_degree_to_direction(270) == "西"
assert skill._wind_degree_to_direction(45) == "东北"
class TestHealthCheck:
@patch.object(WeatherBriefSkill, "_fetch_weather",
return_value=MOCK_API_RESPONSE)
@patch.object(WeatherBriefSkill, "_get_api_key",
return_value="fake-key")
def test_health_ok(self, mock_key, mock_fetch, skill):
result = skill.health_check()
assert result["status"] == "ok"
@patch.object(WeatherBriefSkill, "_get_api_key", return_value="")
def test_health_no_key(self, mock_key, skill):
result = skill.health_check()
assert result["status"] == "error"
assert "API Key" in result["reason"]
# 运行测试
pytest skills/weather_brief/tests/ -v
# 输出示例：
# test_basic_query PASSED
# test_fahrenheit_conversion PASSED
# test_empty_city_raises PASSED
# test_unknown_action_raises PASSED
# test_cache_hit PASSED
# test_cache_expiry PASSED
# test_template_casual PASSED
# test_template_brief PASSED
# test_wind_direction PASSED
# test_health_ok PASSED
# test_health_no_key PASSED
# ======================== 11 passed in 0.43s ========================
第四步：注册并加载 Skill
在 openclaw.config.yaml 中声明
# openclaw.config.yaml
skills:
# 内置 Skill
- email_manager
- calendar_manager
- document_tool
- report_generator
# 自定义 Skill（加在这里）
- weather_brief # ← 新增
# Skill 搜索路径（框架会在这些目录下自动发现 Skill）
skill_paths:
- ./skills # 你的自定义 Skill 目录
- ~/.openclaw/skills # 用户全局 Skill 目录
配置 API Key
# openclaw.secrets.yaml（不要提交到 Git！）
weather_brief:
api_key: your_openweathermap_key_here
# 添加到 .gitignore
echo "openclaw.secrets.yaml" >> .gitignore
验证加载
# verify_skill.py
from openclaw import OpenClawApp
app = OpenClawApp()
app.load_skills()
# 检查 Skill 是否成功加载
skill_info = app.skill_registry.get("weather_brief")
print(f"Skill 名称：{skill_info.name}")
print(f"Skill 状态：{skill_info.health_check()}")
# 直接调用测试
result = app.skill_registry.execute(
skill="weather_brief",
action="generate_brief",
params={"city": "上海", "style": "casual"}
)
print(f"\n天气播报：{result['brief']}")
python verify_skill.py
# 输出：
# ✅ WeatherBriefSkill 初始化成功，API 连接正常（测试城市：Beijing）
# Skill 名称：weather_brief
# Skill 状态：{'status': 'ok', 'cache_entries': 0, 'initialized': True}
#
# 天气播报：今天上海多云，温度 18.5°C，短袖或薄外套即可，出门前看看天哦～
第五步：接入工作流——让 Agent 自动调用
Skill 注册完，Agent 就能在对话中自动调用它了。
对话触发
# Agent 对话示例（无需修改代码，框架自动路由）
agent = app.get_agent("morning_briefing")
# 用户提问
response = agent.chat("今天北京天气怎么样，要带伞吗？")
# → Agent 自动识别意图 → 路由到 weather_brief.get_weather
# → 返回：「今天北京晴，气温 15°C，不用带伞，外套就够了～」
response = agent.chat("帮我生成今天的晨间简报，带上天气信息")
# → Agent 调用多个 Skill 协作：
# calendar_manager.get_events + weather_brief.generate_brief
# → 返回包含天气的完整简报
接入晨间工作流
# workflows/monday_morning_workflow.py（在前几篇基础上扩展）
from openclaw.workflow import Workflow, WorkflowStep
class MondayMorningWorkflow(Workflow):
name = "monday_morning_briefing"
schedule = "0 8 * * 1"
steps = [
WorkflowStep(
name="calendar_briefing",
agent="ScheduleAgent",
action="generate_daily_briefing",
params={"days_ahead": 5}
),
WorkflowStep(
name="email_triage",
agent="EmailClassifierAgent",
action="run",
params={"max_emails": 50}
),
# ← 新增：天气播报步骤
WorkflowStep(
name="weather_brief",
skill="weather_brief", # 直接调用 Skill（不经过 Agent）
action="generate_brief",
params={
"city": "{{user.default_city}}", # 读取用户配置的城市
"style": "casual"
}
),
WorkflowStep(
name="weekly_report",
agent="WeeklyReportGenerator",
action="generate",
depends_on=[]
),
WorkflowStep(
name="push_notification",
agent="NotificationAgent",
action="push_to_wecom",
input_from=[
"calendar_briefing",
"email_triage",
"weather_brief", # ← 天气信息也注入推送
"weekly_report"
]
),
]
实际推送效果
📅 2026年3月25日 周三 晨间简报
🌤 天气
今天北京晴，气温 10~22°C，东北风，紫外线较强。
短袖加外套刚好，记得带防晒～
📆 今日日程（3个）
• 09:30 工程部周同步（会议室B）
• 14:00 与产品确认 Q2 路线图
• 17:00 新员工 1on1（远程）
📬 邮件待处理（2封紧急）
• [紧急] 线上告警：支付服务响应超时
• [待操作] Q1 绩效表格请于本周五前提交
📊 上周数据速览
总销售额 ↑12.3% | 新增用户 ↑18% | 服务可用率 99.94%
开发 Skill 的常见坑
坑 1：description 写得太简单
# ❌ 太简单，Agent 路由时经常选错 Skill
description = "天气查询"
# ✅ 描述清楚触发场景，Agent 才能准确路由
description = """
查询实时天气信息并生成自然语言播报。
适用场景：
- 用户询问某城市今天/明天天气
- 每日简报中自动插入天气信息
- 出行建议（是否需要带伞、穿什么）
"""
坑 2：setup() 里做了太重的初始化
# ❌ setup 里预加载大量数据，导致 Skill 启动慢
def setup(self):
self.all_city_data = self._load_10mb_city_database() # 10MB 数据
# ✅ 懒加载，用到时再加载
def _get_city_data(self):
if not hasattr(self, "_city_data"):
self._city_data = self._load_10mb_city_database()
return self._city_data
坑 3：没处理 API 超时，工作流被卡死
# ❌ 没有超时控制
resp = requests.get(url)
# ✅ 设置超时 + 重试
resp = requests.get(url, timeout=10)
# 加上指数退避重试逻辑（参考 skill.py 中的实现）
坑 4：缓存 Key 没有包含影响结果的所有参数
# ❌ 不同 unit 的结果被混用
cache_key = city
# ✅ 所有影响结果的参数都要进 Key
cache_key = f"{city}:{unit}"
坑 5：直接在 Skill 里写死 API Key
# ❌ 危险！会进 Git
API_KEY = "sk-xxxxxxxx"
# ✅ 通过 secrets 管理器读取（参考安全篇）
api_key = self.secrets.get("weather_brief.api_key")
你的 Skill 发布给别人用
写好的 Skill，可以打包分享给团队或社区。
# 打包
cd skills/weather_brief
zip -r weather_brief_skill_v1.0.zip . \
--exclude "tests/*" \
--exclude "__pycache__/*" \
--exclude "*.pyc"
# 别人安装你的 Skill
openclaw skill install weather_brief_skill_v1.0.zip
# 或者发布到 OpenClaw Skill Hub（社区仓库，规划中）
openclaw skill publish --name weather_brief --version 1.0.0
1 小时回顾
5 min 理解 Skill 和 MCPTool 的区别
10 min 读懂 BaseSkill 骨架
10 min 写 schema.json 和 config.yaml
25 min 实现 skill.py（fetch → normalize → cache → brief）
10 min 写测试，11 个用例全过
5 min 注册、验证、接入工作流
总计：65 分钟（给你 5 分钟刷一杯咖啡）
你现在已经掌握了 OpenClaw Skill 开发的核心流程。所有 Skill 都是这个模式的变体——不管是查天气、读数据库、调内部 API 还是控制 IoT 设备，骨架是一样的。
小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地，聚焦技能开发、插件实践与部署教程，为开发者提供可直接落地的方案、工具与交流平台，助力高效构建与落地 AI 应用
更多推荐
13
0- 0
已为社区贡献19条内容
所有评论(0)