OpenClaw实战：如何通过自定义skill.md文件调用外部API

2次阅读

共计 3000 个字符，预计需要花费 8 分钟才能阅读完成。

OpenClaw 是一个面向开发者的自动化工作流平台，其核心是通过 skill 机制来扩展功能。每个 skill 都是一个独立的功能模块，可以执行特定任务，比如调用外部 API、处理数据等。skill.md文件是定义 skill 的元数据文件，它描述了 skill 的名称、输入输出、依赖关系等信息。

OpenClaw 的基本架构包括：

Skill Engine：负责加载和执行skill
API Gateway：处理外部请求和路由
Scheduler：管理任务调度
Storage：存储 skill 配置和运行数据

在集成外部 API 时，开发者常遇到以下几个挑战：

认证问题：大多数 API 需要 OAuth、JWT 或 API Key 等认证方式
限流处理：API 通常有调用频率限制
数据格式转换：外部 API 返回的数据格式可能与 OpenClaw 预期的格式不一致
错误处理：网络波动、API 变更等不可控因素

skill.md文件采用 YAML 格式，基本结构如下：

name: weather_api
version: 1.0.0
description: Fetches weather data from external API
inputs:
  - name: city
    type: string
    required: true
outputs:
  - name: temperature
    type: float
  - name: conditions
    type: string
endpoints:
  - url: https://api.weather.com/v1
    method: POST
    auth:
      type: oauth2
      config:
        client_id: ${env:WEATHER_CLIENT_ID}
        client_secret: ${env:WEATHER_CLIENT_SECRET}

以下是一个带有 OAuth 认证和错误处理的 Python 实现：

import requests
from typing import Dict, Any
from requests.exceptions import RequestException

class WeatherAPI:
    def __init__(self, client_id: str, client_secret: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = "https://api.weather.com/oauth2/token"
        self.api_url = "https://api.weather.com/v1"
        self.access_token = None

    def _get_access_token(self) -> str:
        """获取 OAuth 访问令牌"""
        try:
            response = requests.post(
                self.token_url,
                data={
                    "grant_type": "client_credentials",
                    "client_id": self.client_id,
                    "client_secret": self.client_secret
                },
                timeout=10
            )
            response.raise_for_status()
            return response.json()["access_token"]
        except RequestException as e:
            raise Exception(f"获取访问令牌失败: {str(e)}")

    def get_weather(self, city: str) -> Dict[str, Any]:
        """获取城市天气数据"""
        if not self.access_token:
            self.access_token = self._get_access_token()

        try:
            response = requests.get(f"{self.api_url}/weather",
                params={"city": city},
                headers={"Authorization": f"Bearer {self.access_token}"},
                timeout=15
            )
            response.raise_for_status()
            return self._transform_data(response.json())
        except RequestException as e:
            # 令牌可能过期，尝试刷新一次
            if response.status_code == 401:
                self.access_token = self._get_access_token()
                return self.get_weather(city)
            raise Exception(f"获取天气数据失败: {str(e)}")

    def _transform_data(self, raw_data: Dict[str, Any]) -> Dict[str, Any]:
        """转换数据格式"""
        return {"temperature": raw_data["main"]["temp"],
            "conditions": raw_data["weather"][0]["description"]
        }

保持一致性 ：确保输出数据结构与skill.md 中定义的 outputs 一致
数据验证：检查 API 返回的数据是否包含所有必要字段
错误处理：提供有意义的错误信息
类型转换：将 API 返回的数据转换为正确的类型

# skill_weather.py
from typing import Dict, Any
import os
from fastapi import FastAPI, HTTPException

app = FastAPI()

# 从环境变量获取认证信息
CLIENT_ID = os.getenv("WEATHER_CLIENT_ID")
CLIENT_SECRET = os.getenv("WEATHER_CLIENT_SECRET")

weather_api = WeatherAPI(CLIENT_ID, CLIENT_SECRET)

@app.post("/weather")
async def get_weather(city: str) -> Dict[str, Any]:
    try:
        return weather_api.get_weather(city)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))