ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

基于 OpenClaw 的 Weather Query Skill 开发实战:用 TaoToken 统一 Key 打通天气查询链路

基于 OpenClaw 的 Weather Query Skill 开发实战:用 TaoToken 统一 Key 打通天气查询链路 1. 为什么要在 OpenClaw 里手搓一个 Weather Query SkillOpenClaw 是一个开源 AI 助手框架它的核心玩法是把通用大模型装进一个可扩展的壳子里再通过 Skills技能把具体业务能力挂上去。你可以把它理解成给模型配了一套工具箱模型负责理解你说的话Skill 负责真正去干活。天气查询就是最典型的入门 Skill——它足够简单能让你把「参数校验 → 异步请求 → 结果解析 → 自然语言格式化」这条链路完整跑一遍又足够实用跑通之后每天都能用。但真正落地时很多人会卡在两个地方。第一是异步请求写得不对aiohttp 的 session 生命周期没管好一并发就报Unclosed client session第二是模型调用的凭证散落在各个 Skill 里今天天气 Skill 写一个 Key明天日程 Skill 又写一个改起来满仓库找。这篇就围绕 OpenClaw 的 Weather Query Skill用 Python 异步调用 Open-Meteo 拿实时天气同时用 TaoToken 把模型调用的 Key 和 API 通道统一收口让你只维护一份凭证配置。适合谁看已经装好 OpenClaw、想写第一个自定义 Skill 的开发者被多个 Skill 各自管 Key 搞烦、想统一凭证的人以及想找一个完整异步 HTTP 实战案例的 Python 学习者。下面所有代码和配置都可以直接复制改掉城市坐标就能跑。2. TaoToken 前置把模型调用凭证统一收口2.1 为什么天气 Skill 也需要模型凭证你可能会问Open-Meteo 是免费 API不需要 Key那天气 Skill 跟模型凭证有什么关系关键在于 OpenClaw 的 Skill 不是孤立运行的。当用户说「帮我看看明天北京冷不冷顺便提醒我带不带伞」框架会先让模型解析意图、抽取location和days参数再路由到 Weather Query Skill。这一步意图解析就是模型调用需要凭证。如果你的 OpenClaw 里挂了五六个 Skill每个都自己读一份 Key维护成本会指数级上升。TaoToken 在这里的角色是统一 Key/API 通道管理你在一处配置好凭证所有 Skill 通过同一套通道去调模型换 Key 只改一个地方。它的 API 地址是https://taotoken.net/api控制台和文档都在官网可查。2.2 拿到统一 Key 并写进环境变量先去控制台创建 API Key然后不要硬编码进代码用环境变量注入# Linux / macOS写进 ~/.bashrc 或 ~/.zshrc export TAOTOKEN_API_KEYsk-你的统一Key export TAOTOKEN_BASE_URLhttps://taotoken.net/api # Windows PowerShell $env:TAOTOKEN_API_KEYsk-你的统一Key $env:TAOTOKEN_BASE_URLhttps://taotoken.net/api注意Key 只放环境变量或密钥管理服务别提交到 Git。OpenClaw 的 Skill 目录经常被整个复制来复制去硬编码的 Key 很容易跟着泄露。2.3 在 OpenClaw 里声明统一通道OpenClaw 的 Skill 通过config.toml读取全局配置。把模型通道写成一份共享配置天气 Skill 只引用不重复定义# ~/.openclaw/config.toml [model] provider taotoken base_url https://taotoken.net/api api_key_env TAOTOKEN_API_KEY default_model claude-sonnet-4-5 [skills.weather_query] enabled true timeout_seconds 10 max_retry 2这样天气 Skill 的代码里只出现config.model.base_url不出现任何 Key 字面量。后面你要加日程 Skill、翻译 Skill全部复用[model]这一段。3. 可复制配置Skill 骨架与异步请求代码3.1 项目结构与 settings.json先建目录结构保持扁平方便 OpenClaw 扫描weather_query/ ├── __init__.py ├── base.py # Skill 基类 ├── skill.py # 天气查询主实现 ├── SKILL.md # 元数据文档 ├── settings.json # Skill 级配置 └── tests/ └── test_skill.pysettings.json放 Skill 自己的运行参数和全局config.toml分离{ skill_name: weather_query, version: 1.0.0, api_endpoint: https://api.open-meteo.com/v1/forecast, timeout_seconds: 10, max_retry: 2, cache_ttl_seconds: 600, default_location: 北京, supported_days: [1, 3, 7] }3.2 Skill 基类定义统一返回结构基类负责把「成功/失败/部分成功」标准化这样 OpenClaw 路由层不用关心每个 Skill 的内部细节# weather_query/base.py from dataclasses import dataclass from enum import Enum from typing import Any, Optional class SkillStatus(Enum): SUCCESS success ERROR error PARTIAL partial dataclass class SkillResult: status: SkillStatus data: Optional[dict] None error: Optional[str] None error_code: Optional[str] None retryable: bool False class BaseSkill: name base_skill description 基础技能 version 1.0.0 triggers: list [] parameters: list [] async def validate(self, args: dict) - tuple: return True, None async def execute(self, **kwargs) - SkillResult: raise NotImplementedError def to_metadata(self) - dict: return { name: self.name, description: self.description, version: self.version, triggers: self.triggers, parameters: self.parameters, }3.3 异步请求核心aiohttp 调 Open-MeteoOpen-Meteo 的 forecast 接口支持current和daily两组参数一次请求就能拿到当前温度和未来几天的最低温/最高温。关键是timezoneAsia/Shanghai否则返回的是 UTC 时间日期会错位# weather_query/skill.py import asyncio import logging import aiohttp from .base import BaseSkill, SkillResult, SkillStatus logger logging.getLogger(__name__) CITY_COORDINATES { 北京: (39.9042, 116.4074), 上海: (31.2304, 121.4737), 广州: (23.1291, 113.2644), 深圳: (22.5431, 114.0579), 呼和浩特: (40.8414, 111.7519), } class WeatherQuerySkill(BaseSkill): name weather_query description 查询指定地区的当前天气和短期预报 version 1.0.0 triggers [天气, 气温, temperature, weather, forecast] parameters [ {name: location, type: string, required: True, description: 城市名称}, {name: days, type: number, required: False, default: 1, description: 预报天数 1-7}, ] API_TIMEOUT_SECONDS 10 MAX_RETRY_ATTEMPTS 2 async def validate(self, args: dict) - tuple: location str(args.get(location, )).strip() if not location: return False, 地点参数不能为空 if len(location) 100: return False, 地点名称过长 days args.get(days, 1) if not isinstance(days, int) or days 1 or days 7: return False, 预报天数必须在 1-7 天之间 return True, None async def execute(self, location: str, days: int 1, **kwargs) - SkillResult: is_valid, error_msg await self.validate( {location: location, days: days} ) if not is_valid: return SkillResult( statusSkillStatus.ERROR, errorerror_msg, error_codeINVALID_PARAMS, ) for attempt in range(self.MAX_RETRY_ATTEMPTS): try: return await self._fetch_weather(location, days) except asyncio.TimeoutError: if attempt self.MAX_RETRY_ATTEMPTS - 1: await asyncio.sleep(2) except aiohttp.ClientError as e: return SkillResult( statusSkillStatus.ERROR, errorf网络请求失败{e}, error_codeNETWORK_ERROR, retryableTrue, ) return SkillResult( statusSkillStatus.ERROR, error天气查询超时, error_codeTIMEOUT, retryableTrue, ) def _get_coordinates(self, location: str) - tuple: for city, coords in CITY_COORDINATES.items(): if city in location: return coords return CITY_COORDINATES[北京] async def _fetch_weather(self, location: str, days: int) - SkillResult: lat, lon self._get_coordinates(location) url ( https://api.open-meteo.com/v1/forecast f?latitude{lat}longitude{lon} currenttemperature_2m,relative_humidity_2m,weather_code dailytemperature_2m_max,temperature_2m_min fforecast_days{min(days, 7)} timezoneAsia%2FShanghai ) timeout aiohttp.ClientTimeout(totalself.API_TIMEOUT_SECONDS) async with aiohttp.ClientSession(timeouttimeout) as session: async with session.get(url) as response: if response.status ! 200: return SkillResult( statusSkillStatus.ERROR, errorfAPI 返回状态码 {response.status}, error_codeAPI_ERROR, ) data await response.json() return SkillResult( statusSkillStatus.SUCCESS, dataself._parse_api_response(data, location, days), ) def _parse_api_response(self, data: dict, location: str, days: int) - dict: current data.get(current, {}) daily data.get(daily, {}) code current.get(weather_code, 0) condition 晴朗 if code 3 else (阴天 if code 50 else 有降水) forecast [] for i in range(min(days, len(daily.get(time, [])))): forecast.append({ date: daily[time][i], max_temp_c: daily[temperature_2m_max][i], min_temp_c: daily[temperature_2m_min][i], }) return { location: location, current: { temperature_c: current.get(temperature_2m), humidity: current.get(relative_humidity_2m), condition: condition, }, forecast: forecast, } def format_response(self, result: SkillResult) - str: if result.status ! SkillStatus.SUCCESS: return f查询失败{result.error} data result.data cur data[current] lines [ f{data[location]}当前温度 {cur[temperature_c]}°C f{cur[condition]}湿度 {cur[humidity]}% ] for day in data[forecast]: lines.append( f{day[date]}{day[min_temp_c]}°C ~ {day[max_temp_c]}°C ) return \n.join(lines)这里有个容易踩的坑aiohttp.ClientSession必须用async with包住否则连接池不会释放跑几十次之后就会看到Unclosed client session警告。我试过把 session 提到类属性复用结果在并发场景下反而更容易出问题所以每个请求独立开 session 是最稳的写法。4. 验证请求从本地触发到结果校验4.1 本地直接跑一次不用等 OpenClaw 加载先写个最小脚本验证 Skill 本身能跑通# test_local.py import asyncio from weather_query.skill import WeatherQuerySkill async def main(): skill WeatherQuerySkill() result await skill.execute(location呼和浩特, days3) print(skill.format_response(result)) asyncio.run(main())预期输出类似呼和浩特当前温度 7.2°C阴天湿度 22% 2026-03-10-0.5°C ~ 9.6°C 2026-03-11-1.2°C ~ 10.3°C 2026-03-120.4°C ~ 11.8°C如果温度是None说明current参数没生效检查 URL 里current后面的字段名有没有拼错。Open-Meteo 对字段名很严格写错不会报错只会静默返回空。4.2 挂进 OpenClaw 并触发把整个weather_query目录复制到 OpenClaw 的 skills 目录cp -r ./weather_query ~/.openclaw/skills/ ls ~/.openclaw/skills/ | grep weather然后在 OpenClaw 对话里输入「北京今天天气怎么样」框架会走「模型解析意图 → 路由到 weather_query → 执行 → 格式化」这条链路。如果模型没路由过去检查triggers里有没有覆盖用户可能说的词中文用户经常说「冷不冷」「要不要带伞」这些可以补进 triggers。4.3 用 curl 单独验证 Open-Meteo 链路当 Skill 报错但你不确定是代码问题还是 API 问题时直接用 curl 打一次原始接口把变量隔离出来curl -s https://api.open-meteo.com/v1/forecast?latitude39.9042longitude116.4074currenttemperature_2m,relative_humidity_2m,weather_codedailytemperature_2m_max,temperature_2m_minforecast_days1timezoneAsia%2FShanghai返回的 JSON 里current.temperature_2m有值就说明 API 正常问题在 Python 侧如果 curl 也空那就是参数拼错了。5. 本篇常见错排查5.1 Unclosed client session 警告现象跑几次之后控制台刷Unclosed client session。原因基本是ClientSession没用async with或者异常路径下 session 没关闭。修法就是像上面代码那样把 session 和 response 都用async with包住异常交给外层 try 处理。5.2 日期错位一天现象返回的预报日期比实际早一天。这是时区问题Open-Meteo 默认 UTC必须显式传timezoneAsia/Shanghai。注意 URL 里斜杠要编码成%2F直接写Asia/Shanghai有些 HTTP 客户端会截断。5.3 模型不触发 Skill现象用户说天气模型却自己编了一段回答没走 Skill。两个原因一是triggers覆盖不够补上口语化说法二是 Skill 的to_metadata()没被 OpenClaw 读到检查SKILL.md是否存在、__init__.py是否导出了 Skill 类。5.4 统一 Key 读取失败现象Skill 能查天气但意图解析阶段报鉴权错误。检查TAOTOKEN_API_KEY是否在当前 shell 生效echo $TAOTOKEN_API_KEY看有没有值。如果你是在 systemd 或 Docker 里跑 OpenClaw环境变量不会自动继承要在服务配置里显式声明。5.5 并发请求被限流Open-Meteo 免费额度大约每分钟 60 次单机跑没问题但如果你的 OpenClaw 同时被多个用户触发建议加一层内存缓存_cache {} _cache_ttl 600 async def _fetch_weather_cached(self, location: str, days: int): key f{location}_{days} now asyncio.get_event_loop().time() if key in _cache and now - _cache[key][ts] _cache_ttl: return _cache[key][result] result await self._fetch_weather(location, days) _cache[key] {ts: now, result: result} return result6. 把凭证和 Skill 一起管起来天气 Skill 跑通之后你会发现真正省事的地方在于模型调用的 Key 只在config.toml里出现一次天气 Skill、日程 Skill、翻译 Skill 全部复用同一条通道。以后换模型、换额度改一处就行不用满仓库搜sk-。如果你还在调 Skill 的意图解析想先确认模型能不能正确抽取location和days可以直接在模型对话里试几轮把参数抽对了再写进代码。等你要把天气 Skill 和编码类 Skill 串成长期运行的 Agent 工作流Coding Plan 那种按周期计费的方式会比按次调用更可控。接入过程中如果遇到鉴权或通道配置问题接入文档里有各语言的示例API Keys 页面可以随时轮换凭证。
返回列表