OpenClaw高级技巧:自定义技能开发与插件生态

skill-name/
├── SKILL.md          # 技能说明文档
├── config.json       # 配置文件（可选）
├── scripts/          # 执行脚本
│   ├── main.py       # 主逻辑
│   └── utils.py      # 工具函数
├── references/       # 参考文档
└── tests/           # 测试文件

---
name: skill-name
description: 技能描述
---

# 技能名称

## 何时使用
- 场景1：当用户需要...
- 场景2：当用户要求...

## 工作流
1. 第一步：...
2. 第二步：...
3. 第三步：...

## 配置说明
- 参数1：说明
- 参数2：说明

## 示例
```bash
# 使用示例

## 🚀 实战：创建你的第一个技能

### 案例：天气查询技能

#### 步骤1：创建技能目录结构
```bash
mkdir -p weather-skill/{scripts,references}
cd weather-skill

---
name: weather-skill
description: 查询指定城市的天气信息
---

# 天气查询技能

## 何时使用
- 当用户询问某个城市的天气时
- 当用户需要出行天气建议时

## 工作流
1. 解析用户输入，提取城市名称
2. 调用天气API获取数据
3. 格式化返回结果
4. 提供出行建议

## 配置说明
- API_KEY：天气API密钥（可选）
- DEFAULT_CITY：默认城市（可选）

## 示例
```bash
# 查询北京天气
weather 北京

# 查询上海天气（带建议）
weather 上海 --advice

#### 步骤3：编写主脚本
```python
# scripts/main.py
import sys
import requests
import json

def get_weather(city, api_key=None):
    """获取城市天气"""
    # 这里使用公开的天气API示例
    url = f"https://wttr.in/{city}?format=j1"

    try:
        response = requests.get(url, timeout=10)
        data = response.json()

        # 解析天气数据
        current = data['current_condition'][0]
        weather_info = {
            'city': city,
            'temperature': current['temp_C'],
            'condition': current['weatherDesc'][0]['value'],
            'humidity': current['humidity'],
            'wind_speed': current['windspeedKmph']
        }

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

def format_output(weather_info, include_advice=False):
    """格式化输出"""
    if 'error' in weather_info:
        return f"❌ 获取天气失败: {weather_info['error']}"

    output = [
        f"🌤️  {weather_info['city']} 天气报告",
        f"🌡️  温度: {weather_info['temperature']}°C",
        f"☁️  天气: {weather_info['condition']}",
        f"💧 湿度: {weather_info['humidity']}%",
        f"💨 风速: {weather_info['wind_speed']} km/h"
    ]

    if include_advice:
        advice = generate_advice(weather_info)
        output.append(f"\n💡 出行建议: {advice}")

    return "\n".join(output)

def generate_advice(weather_info):
    """生成出行建议"""
    temp = int(weather_info['temperature'])
    condition = weather_info['condition'].lower()

    if temp < 0:
        return "天气寒冷，请注意保暖，穿厚外套"
    elif temp < 10:
        return "天气较冷，建议穿外套"
    elif temp < 25:
        return "天气舒适，适合户外活动"
    else:
        return "天气炎热，注意防暑降温"

    if 'rain' in condition:
        return "有雨，请带伞"
    elif 'snow' in condition:
        return "下雪，注意防滑"

    return "天气适宜，出行愉快"

if __name__ == "__main__":
    # 解析命令行参数
    if len(sys.argv) < 2:
        print("用法: python main.py <城市> [--advice]")
        sys.exit(1)

    city = sys.argv[1]
    include_advice = '--advice' in sys.argv

    weather = get_weather(city)
    result = format_output(weather, include_advice)
    print(result)

# 测试基本功能
python scripts/main.py 北京

# 测试带建议的功能
python scripts/main.py 上海 --advice

import argparse

def parse_args():
    parser = argparse.ArgumentParser(description='天气查询技能')
    parser.add_argument('city', help='城市名称')
    parser.add_argument('--advice', action='store_true', help='包含出行建议')
    parser.add_argument('--api-key', help='API密钥')
    parser.add_argument('--format', choices=['text', 'json', 'html'],
                       default='text', help='输出格式')
    return parser.parse_args()

import time
from functools import wraps

def retry(max_attempts=3, delay=1):
    """重试装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise
                    time.sleep(delay)
            return None
        return wrapper
    return decorator

@retry(max_attempts=3, delay=2)
def call_external_api(url):
    """调用外部API（带重试）"""
    response = requests.get(url, timeout=10)
    response.raise_for_status()
    return response.json()

import hashlib
import pickle
from datetime import datetime, timedelta

class Cache:
    def __init__(self, ttl_minutes=30):
        self.cache = {}
        self.ttl = timedelta(minutes=ttl_minutes)

    def get_key(self, *args, **kwargs):
        """生成缓存键"""
        data = str(args) + str(sorted(kwargs.items()))
        return hashlib.md5(data.encode()).hexdigest()

    def get(self, key):
        """获取缓存"""
        if key in self.cache:
            value, timestamp = self.cache[key]
            if datetime.now() - timestamp < self.ttl:
                return value
        return None

    def set(self, key, value):
        """设置缓存"""
        self.cache[key] = (value, datetime.now())

import asyncio
import aiohttp

async def fetch_multiple_cities(cities):
    """异步获取多个城市天气"""
    async with aiohttp.ClientSession() as session:
        tasks = []
        for city in cities:
            task = fetch_city_weather(session, city)
            tasks.append(task)

        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results

async def fetch_city_weather(session, city):
    """获取单个城市天气"""
    url = f"https://wttr.in/{city}?format=j1"
    async with session.get(url) as response:
        data = await response.json()
        return process_weather_data(data, city)

用户请求 → OpenClaw网关 → 插件路由器 → 技能执行器 → 结果返回
                ↓
          插件管理器
                ↓
        技能注册与发现

# 插件注册示例
from openclaw.plugin import register_skill

@register_skill(
    name="weather",
    description="查询天气信息",
    version="1.0.0",
    author="Your Name"
)
class WeatherSkill:
    def __init__(self, config):
        self.config = config
        self.api_key = config.get('api_key')

    async def execute(self, context):
        """执行技能"""
        city = context.get('city')
        result = await self.get_weather(city)
        return self.format_result(result)

    async def get_weather(self, city):
        # 获取天气逻辑
        pass

import importlib
import pkgutil
from pathlib import Path

def discover_skills(skills_dir):
    """发现技能目录中的所有技能"""
    skills = {}

    for skill_dir in Path(skills_dir).iterdir():
        if skill_dir.is_dir() and (skill_dir / "SKILL.md").exists():
            skill_name = skill_dir.name
            try:
                # 动态加载技能模块
                module_path = f"skills.{skill_name}.main"
                module = importlib.import_module(module_path)
                skills[skill_name] = module
            except ImportError as e:
                print(f"无法加载技能 {skill_name}: {e}")

    return skills

import click

@click.group()
def cli():
    """OpenClaw技能管理CLI"""
    pass

@cli.command()
@click.argument('city')
@click.option('--advice', is_flag=True, help='包含出行建议')
def weather(city, advice):
    """查询城市天气"""
    result = get_weather_with_advice(city, advice)
    click.echo(result)

@cli.command()
def list_skills():
    """列出所有可用技能"""
    skills = discover_skills('./skills')
    for name, module in skills.items():
        click.echo(f"• {name}: {module.__doc__ or '无描述'}")

if __name__ == '__main__':
    cli()

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/api/skills/weather', methods=['POST'])
def weather_api():
    """天气查询API"""
    data = request.json
    city = data.get('city', '北京')

    weather_info = get_weather(city)
    return jsonify({
        'success': 'error' not in weather_info,
        'data': weather_info
    })

@app.route('/api/skills', methods=['GET'])
def list_skills_api():
    """列出技能API"""
    skills = discover_skills('./skills')
    return jsonify({
        'skills': list(skills.keys()),
        'count': len(skills)
    })

from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes

async def weather_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Telegram天气命令"""
    if not context.args:
        await update.message.reply_text("请指定城市，例如：/weather 北京")
        return

    city = ' '.join(context.args)
    weather_info = get_weather(city)

    if 'error' in weather_info:
        await update.message.reply_text(f"获取天气失败: {weather_info['error']}")
    else:
        message = format_weather_message(weather_info)
        await update.message.reply_text(message)

def setup_telegram_bot():
    """设置Telegram机器人"""
    application = Application.builder().token("YOUR_BOT_TOKEN").build()

    # 注册技能命令
    application.add_handler(CommandHandler("weather", weather_command))

    return application

# 创建技能包结构
weather-skill-package/
├── setup.py
├── README.md
├── skill/
│   ├── __init__.py
│   ├── main.py
│   └── utils.py
└── requirements.txt

from setuptools import setup, find_packages

setup(
    name="openclaw-weather-skill",
    version="1.0.0",
    description="OpenClaw天气查询技能",
    author="Your Name",
    packages=find_packages(),
    install_requires=[
        'requests>=2.25.0',
        'click>=8.0.0',
    ],
    entry_points={
        'openclaw.skills': [
            'weather = skill.main:WeatherSkill',
        ],
    },
    classifiers=[
        'Development Status :: 4 - Beta',
        'Intended Audience :: Developers',
        'Topic :: Software Development :: Libraries',
    ],
)

# 构建包
python setup.py sdist bdist_wheel

# 发布到PyPI
twine upload dist/*

# 或发布到私有仓库
twine upload --repository-url https://your-private-repo.com dist/*

import unittest
from skill.main import get_weather, format_output

class TestWeatherSkill(unittest.TestCase):
    def test_get_weather_success(self):
        """测试成功获取天气"""
        result = get_weather('北京')
        self.assertIn('temperature', result)
        self.assertIn('condition', result)

    def test_format_output(self):
        """测试格式化输出"""
        test_data = {
            'city': '测试城市',
            'temperature': '20',
            'condition': '晴朗',
            'humidity': '50',
            'wind_speed': '10'
        }

        output = format_output(test_data)
        self.assertIn('测试城市', output)
        self.assertIn('20°C', output)

    def test_error_handling(self):
        """测试错误处理"""
        result = get_weather('不存在的城市')
        self.assertIn('error', result)

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

import pytest
from openclaw.testing import SkillTestClient

@pytest.fixture
def weather_skill():
    """创建天气技能实例"""
    from skill.main import WeatherSkill
    return WeatherSkill(config={'api_key': 'test_key'})

def test_weather_skill_execution(weather_skill):
    """测试技能执行"""
    context = {'city': '北京'}
    result = weather_skill.execute(context)

    assert 'temperature' in result
    assert 'condition' in result
    assert result['city'] == '北京'

def test_weather_skill_with_advice(weather_skill):
    """测试带建议的技能执行"""
    context = {'city': '北京', 'advice': True}
    result = weather_skill.execute(context)

    assert 'advice' in result
    assert isinstance(result['advice'], str)

import time
import statistics

def benchmark_skill(skill_func, iterations=100):
    """性能基准测试"""
    times = []

    for i in range(iterations):
        start = time.time()
        skill_func()
        end = time.time()
        times.append(end - start)

    return {
        'min': min(times),
        'max': max(times),
        'mean': statistics.mean(times),
        'median': statistics.median(times),
        'p95': statistics.quantiles(times, n=20)[18]  # 95th percentile
    }

# 运行性能测试
results = benchmark_weather_skill(lambda: get_weather('北京'))
print(f"性能指标: {results}")

categories:
  productivity:    # 生产力工具
    - task-management
    - note-taking
    - calendar
  communication:   # 通信工具
    - email
    - messaging
    - video-call
  development:     # 开发工具
    - code-review
    - deployment
    - monitoring
  entertainment:   # 娱乐工具
    - music
    - games
    - news
  utilities:       # 实用工具
    - weather
    - translation
    - calculator

class SkillStore:
    def __init__(self):
        self.skills = {}
        self.categories = {}
        self.ratings = {}

    def register_skill(self, skill_metadata):
        """注册技能到商店"""
        skill_id = skill_metadata['id']
        self.skills[skill_id] = skill_metadata

        # 分类管理
        for category in skill_metadata.get('categories', []):
            if category not in self.categories:
                self.categories[category] = []
            self.categories[category].append(skill_id)

    def search_skills(self, query, category=None):
        """搜索技能"""
        results = []

        for skill_id, skill in self.skills.items():
            # 匹配查询
            matches_query = (
                query.lower() in skill['name'].lower() or
                query.lower() in skill['description'].lower() or
                any(query.lower() in tag.lower() for tag in skill.get('tags', []))
            )

            # 匹配分类
            matches_category = (
                category is None or
                category in skill.get('categories', [])
            )

            if matches_query and matches_category:
                results.append(skill)

        # 排序：评分高的在前
        results.sort(key=lambda x: self.ratings.get(x['id'], 0), reverse=True)
        return results

    def rate_skill(self, skill_id, rating, feedback=None):
        """为技能评分"""
        if skill_id not in self.ratings:
            self.ratings[skill_id] = []

        self.ratings[skill_id].append({
            'rating': rating,
            'feedback': feedback,
            'timestamp': datetime.now()
        })

graph TB
    A[技能开发者] --> B[技能开发平台]
    B --> C[技能测试环境]
    C --> D[技能发布系统]
    D --> E[技能商店]
    E --> F[技能使用者]
    F --> G[使用反馈]
    G --> H[技能优化]
    H --> B

    I[管理员] --> J[审核系统]
    J --> K[质量监控]
    K --> L[安全扫描]
    L --> M[版本管理]