@clawhub-jasonzhang2015-6b6f4d1785
校验模型配置是否正确、模型是否可以正常连接和返回内容。当用户说"检查模型"、"测试模型"、"模型能不能用"、"模型配置"、"诊断模型问题"时使用。**每次修改模型配置(config.patch/config.apply涉及models.providers)后必须自动执行校验。** 用户只给模型名+API key时...
---
name: model-config-check
description: 校验模型配置是否正确、模型是否可以正常连接和返回内容。当用户说"检查模型"、"测试模型"、"模型能不能用"、"模型配置"、"诊断模型问题"时使用。**每次修改模型配置(config.patch/config.apply涉及models.providers)后必须自动执行校验。** 用户只给模型名+API key时,自动识别provider、查找配置、写入并校验。
---
# 模型配置校验 Skill
## ⚡ 自动触发规则
**当以下情况发生时,必须自动运行校验脚本,无需用户请求:**
1. 使用 `gateway config.patch` 修改了 `models` 相关配置(新增/修改 provider 或 model)
2. 使用 `gateway config.apply` 替换了整个配置且包含 models 变更
3. 使用 `gateway update.run` 更新后(模型接口可能有变化)
**自动校验流程:**
1. 配置写入完成 + gateway 重启后,等待 5 秒让服务就绪
2. 执行 `bash ~/.openclaw/workspace/skills/model-config-check/scripts/check_models.sh`
3. 解析输出,向用户汇报校验结果
4. 如果新模型不可用,立即告知用户具体原因和修复建议
**手动触发:** 用户说"检查模型"/"测试模型"/"模型能不能用"/"诊断模型"时也执行同样流程。
## 🤖 自动配置流程
当用户只提供「模型名 + API key」时,按以下流程自动完成配置:
### Step 1: 识别 Provider
根据模型名前缀匹配已知 provider:
| 模型名前缀 | Provider | Base URL | API 类型 |
|-----------|----------|----------|----------|
| `gpt-*`, `o1-*`, `o3-*`, `o4-*`, `chatgpt-*` | OpenAI | `https://api.openai.com` | `openai-completions` → `/v1/chat/completions` |
| `claude-*` | Anthropic | `https://api.anthropic.com` | `anthropic-messages` → `/v1/messages` |
| `deepseek-*` | DeepSeek | `https://api.deepseek.com` | `openai-completions` → `/v1/chat/completions` |
| `glm-*`, `chatglm-*` | 智谱 (Zhipu) | `https://open.bigmodel.cn/api/paas/v4` | `openai-completions` → `/v1/chat/completions` |
| `qwen-*`, `qwq-*`, `qvq-*` | 阿里通义 (DashScope) | `https://dashscope.aliyuncs.com/compatible-mode/v1` | `openai-completions` → `/v1/chat/completions` |
| `moonshot-*`, `kimi-*` | 月之暗面 (Moonshot) | `https://api.moonshot.cn/v1` | `openai-completions` → `/v1/chat/completions` |
| `doubao-*`, `ep-*` | 火山引擎 (豆包) | `https://ark.cn-beijing.volces.com/api/v3` | `openai-completions` → `/v1/chat/completions` |
| `minimax-*`, `abab-*` | MiniMax | `https://api.minimax.chat/v1` | `openai-completions` → `/v1/chat/completions` |
| `yi-*` | 零一万物 (01.AI) | `https://api.lingyiwanwu.com/v1` | `openai-completions` → `/v1/chat/completions` |
| `ernie-*`, `baidu-*` | 百度文心 | `https://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop` | `openai-completions` |
| `grok-*` | xAI (Grok) | `https://api.x.ai/v1` | `openai-completions` → `/v1/chat/completions` |
| `gemini-*` | Google Gemini | `https://generativelanguage.googleapis.com/v1beta` | 特殊接口,需单独处理 |
| `mistral-*`, `codestral-*` | Mistral | `https://api.mistral.ai/v1` | `openai-completions` → `/v1/chat/completions` |
| `mimo-*` | 小米 (MiMo) | `https://api.xiaomimimo.com/anthropic` | `anthropic-messages` → `/v1/messages` |
**未匹配时:** 自动搜索 "[model_name] API documentation base_url" 确认配置。
### Step 2: 生成配置
根据匹配结果生成 config.patch 格式的配置,包含:
- provider 名称(用 provider 小写名作为 key)
- baseUrl
- apiKey(用户提供的)
- api 类型
- models 列表(至少包含用户提到的模型)
- 每个 model 的 contextWindow 和 maxTokens(根据已知信息填写默认值)
### Step 3: 应用配置
使用 `gateway config.patch` 写入配置并重启。
### Step 4: 自动校验
重启完成后执行校验脚本,汇报结果。
### 模型上下文/输出默认值
常见模型的 contextWindow 和 maxTokens 参考值:
| 模型 | contextWindow | maxTokens |
|------|-------------|-----------|
| gpt-4o | 128000 | 16384 |
| gpt-4-turbo | 128000 | 4096 |
| o1/o1-mini | 128000 | 100000 |
| claude-3.5-sonnet | 200000 | 8192 |
| claude-3-opus | 200000 | 4096 |
| deepseek-chat | 64000 | 8192 |
| deepseek-reasoner | 64000 | 8192 |
| glm-4 | 128000 | 4096 |
| qwen-max | 32000 | 8192 |
| qwen-long | 1000000 | 65536 |
| kimi-latest | 128000 | 4096 |
| doubao-pro | 4096 (volcengine) | 4096 |
| minimax-abab6.5 | 245760 | 4096 |
| mimo-v2-pro | 262144 | 8192 |
**未知模型:** contextWindow 默认 128000,maxTokens 默认 4096。搜文档确认后更新。
按以下顺序逐项检查,每项给出 ✅/❌ 结果:
### 1. 读取配置
使用 `read` 工具读取 `~/.openclaw/openclaw.json`,提取所有 `models.providers` 下的模型配置。
### 2. 配置完整性检查
对每个 provider 检查:
- `baseUrl` 是否存在且格式正确(http/https 开头)
- `apiKey` 是否存在且非空
- `api` 类型是否正确(`openai-completions` / `anthropic-messages` / `anthropic-completions`)
- 每个 model 的 `id` 是否存在
- `contextWindow` 和 `maxTokens` 是否设置
### 3. 网络连通性检查
使用 `exec` 执行 curl 测试每个 provider 的 baseUrl 是否可达:
```bash
curl -s -o /dev/null -w "%{http_code}" --connect-timeout 5 "<baseUrl>"
```
### 4. API 实际调用测试
对每个 provider,使用 `exec` 执行实际 API 调用测试:
**OpenAI 兼容接口 (`openai-completions`)**:
```bash
curl -s -X POST "<baseUrl>" \
-H "Content-Type: application/json" \
-H "Authorization: Bearer <apiKey>" \
-d '{"model":"<modelId>","messages":[{"role":"user","content":"reply 1"}],"max_tokens":10}' \
--connect-timeout 10 --max-time 30
```
**Anthropic 兼容接口 (`anthropic-messages`)**:
```bash
curl -s -X POST "<baseUrl>/v1/messages" \
-H "Content-Type: application/json" \
-H "x-api-key: <apiKey>" \
-H "anthropic-version: 2023-06-01" \
-d '{"model":"<modelId>","max_tokens":10,"messages":[{"role":"user","content":"reply 1"}]}' \
--connect-timeout 10 --max-time 30
```
注意:对于 anthropic-messages,baseUrl 通常不带 `/v1/messages`,需要拼接。有些 baseUrl 已经包含路径,则直接用。
### 5. 结果解析
检查返回结果:
- HTTP 状态码是否为 200
- 响应体是否包含有效内容(非空)
- 对于 OpenAI 接口:检查 `choices[0].message.content` 是否非空
- 对于 Anthropic 接口:检查 `content[0].text` 是否非空
- 如果 content 为空但 reasoning_content 有值,说明模型把输出放到了思考字段,需标注配置问题
### 6. 生成报告
汇总输出格式:
```
## 模型配置校验报告
### Provider: <name>
- 配置完整性: ✅/❌
- 网络连通: ✅/❌ (HTTP <code>)
- API 认证: ✅/❌
- 模型返回: ✅/❌
- 模型: <modelId> → 状态: ✅/❌ [备注]
### 总结
- 可用模型: X/Y
- 不可用模型: [列表]
- 建议: [修复建议]
```
## URL 路径处理规则
不同 provider 的 baseUrl 结构不同,脚本自动处理:
| API 类型 | 处理规则 |
|----------|----------|
| `anthropic-messages` | 自动追加 `/v1/messages`(除非 URL 已包含) |
| `openai-completions` | 自动追加 `/chat/completions`(除非 URL 已以 `/chat/completions` 或 `/completions` 结尾) |
## 常见问题及修复
| 问题 | 原因 | 修复方式 |
|------|------|----------|
| HTTP 401/403 | API Key 无效或过期 | 更新 apiKey |
| HTTP 500 | 服务端错误/账户耗尽 | 检查账户余额或联系服务商 |
| 连接超时 | baseUrl 不可达 | 检查网络或更换 baseUrl |
| content 为空 | 接口类型不匹配 | 检查 api 字段是否正确(openai vs anthropic) |
| content 为空但 reasoning 有值 | 模型输出到了思考字段 | 增加 max_tokens 或切换接口类型 |
| Relay service error | 中转服务异常 | 检查中转服务状态和账户 |
| THINKING_ONLY | 推理模型思考未完成 | 正常现象,模型可用,增加 max_tokens 可获取完整输出 |
FILE:scripts/check_models.sh
#!/bin/bash
# check_models.sh - 快速检查所有模型配置和连通性
# 用法: bash check_models.sh [config_path]
CONFIG="-$HOME/.openclaw/openclaw.json"
ISSUES=()
OK_COUNT=0
TOTAL=0
GREEN='\033[0;32m'
RED='\033[0;31m'
YELLOW='\033[1;33m'
NC='\033[0m'
echo "=========================================="
echo " 🔍 OpenClaw 模型配置校验"
echo "=========================================="
echo ""
if [ ! -f "$CONFIG" ]; then
echo -e "RED❌ 配置文件不存在: CONFIGNC"
exit 1
fi
echo "✅ 配置文件: CONFIG"
echo ""
# Extract providers
TMPEXTRACT=$(mktemp)
cat > "$TMPEXTRACT" << 'PYEOF'
import json, sys
config_path = sys.argv[1]
with open(config_path) as f:
config = json.load(f)
providers = config.get("models", {}).get("providers", {})
for name, p in providers.items():
models = [m["id"] for m in p.get("models", [])]
base = p.get("baseUrl", "")
key = p.get("apiKey", "")
api = p.get("api", "")
print(f"{name}\t{base}\t{key}\t{api}\t{','.join(models)}")
PYEOF
PROVIDERS=$(python3 "$TMPEXTRACT" "$CONFIG" 2>/dev/null)
rm -f "$TMPEXTRACT"
if [ -z "$PROVIDERS" ]; then
echo -e "RED❌ 未找到任何模型 provider 配置NC"
exit 1
fi
# Parser for Anthropic responses
TMPPARSE=$(mktemp)
cat > "$TMPPARSE" << 'PYPARSE'
import json, sys
resp_file = sys.argv[1]
with open(resp_file) as f:
raw = f.read().strip()
if not raw:
print("EMPTY")
sys.exit(0)
try:
r = json.loads(raw)
# Check for text content
for c in r.get("content", []):
if c.get("type") == "text" and c.get("text", "").strip():
print(c["text"].strip())
sys.exit(0)
# Check for thinking content (model is reasoning but didn't produce text yet)
for c in r.get("content", []):
if c.get("type") == "thinking" and c.get("thinking", "").strip():
print("THINKING_ONLY: " + c["thinking"].strip()[:80])
sys.exit(0)
if "error" in r:
print("ERROR: " + json.dumps(r["error"]))
else:
print("EMPTY")
except Exception as e:
print("PARSE_FAIL: " + str(e))
PYPARSE
# Parser for OpenAI responses
TMPPARSE_OAI=$(mktemp)
cat > "$TMPPARSE_OAI" << 'PYPARSE'
import json, sys
resp_file = sys.argv[1]
with open(resp_file) as f:
raw = f.read().strip()
if not raw:
print("EMPTY")
sys.exit(0)
try:
r = json.loads(raw)
choices = r.get("choices", [])
if choices:
msg = choices[0].get("message", {})
content = msg.get("content", "")
if content and content.strip():
print(content.strip())
sys.exit(0)
reasoning = msg.get("reasoning_content", "")
if reasoning and reasoning.strip():
print("REASONING_ONLY: " + reasoning.strip()[:80])
sys.exit(0)
if "error" in r:
print("ERROR: " + json.dumps(r["error"]))
else:
print("EMPTY")
except Exception as e:
print("PARSE_FAIL: " + str(e))
PYPARSE
while IFS=$'\t' read -r NAME BASEURL APIKEY API_TYPE MODELS; do
echo "----------------------------------------"
echo "📦 Provider: NAME"
echo " API 类型: API_TYPE"
echo " Base URL: BASEURL"
if [ -z "$BASEURL" ]; then
echo -e " RED❌ baseUrl 为空NC"
ISSUES+=("NAME: baseUrl 为空")
continue
fi
if [ -z "$APIKEY" ]; then
echo -e " RED❌ apiKey 为空NC"
ISSUES+=("NAME: apiKey 为空")
continue
fi
echo -e " GREEN✅ 配置完整NC"
# Network check
HTTP_CODE=$(curl -s -o /dev/null -w "%{http_code}" --connect-timeout 5 "$BASEURL" 2>/dev/null)
if [ "$HTTP_CODE" = "000" ]; then
echo -e " RED❌ 网络不可达NC"
ISSUES+=("NAME: 网络不可达")
continue
fi
echo -e " GREEN✅ 网络连通NC (HTTP HTTP_CODE)"
IFS=',' read -ra MODEL_LIST <<< "$MODELS"
for MODEL_ID in "MODEL_LIST[@]"; do
TOTAL=$((TOTAL + 1))
echo -n " 🧪 MODEL_ID ... "
TMPRESP=$(mktemp)
if [[ "$API_TYPE" == *"anthropic"* ]]; then
# Build URL: append /v1/messages if not already in the path
if [[ "$BASEURL" == *"/v1/messages"* ]]; then
FULL_URL="$BASEURL"
else
FULL_URL="BASEURL%//v1/messages"
fi
curl -s -X POST "$FULL_URL" \
-H "Content-Type: application/json" \
-H "x-api-key: APIKEY" \
-H "anthropic-version: 2023-06-01" \
-d "{\"model\":\"MODEL_ID\",\"max_tokens\":512,\"messages\":[{\"role\":\"user\",\"content\":\"reply ok\"}]}" \
--connect-timeout 10 --max-time 60 \
-o "$TMPRESP" 2>/dev/null
TEXT=$(python3 "$TMPPARSE" "$TMPRESP" 2>/dev/null)
else
# OpenAI: construct the full endpoint URL
# If baseUrl doesn't end with /chat/completions or /completions, append /chat/completions
if [[ "$BASEURL" == *"/chat/completions" ]] || [[ "$BASEURL" == *"/completions" ]]; then
OAI_URL="$BASEURL"
else
OAI_URL="BASEURL%//chat/completions"
fi
curl -s -X POST "$OAI_URL" \
-H "Content-Type: application/json" \
-H "Authorization: Bearer APIKEY" \
-d "{\"model\":\"MODEL_ID\",\"messages\":[{\"role\":\"user\",\"content\":\"reply ok\"}],\"max_tokens\":512}" \
--connect-timeout 10 --max-time 60 \
-o "$TMPRESP" 2>/dev/null
TEXT=$(python3 "$TMPPARSE_OAI" "$TMPRESP" 2>/dev/null)
fi
rm -f "$TMPRESP"
if [[ "$TEXT" == "THINKING_ONLY"* ]] || [[ "$TEXT" == "REASONING_ONLY"* ]]; then
echo -e "YELLOW⚠️ 模型工作正常,输出在 reasoning 字段NC"
OK_COUNT=$((OK_COUNT + 1))
elif [[ "$TEXT" == "ERROR"* ]]; then
echo -e "RED❌ TEXTNC"
ISSUES+=("MODEL_ID: TEXT")
elif [[ "$TEXT" == "EMPTY" ]] || [[ "$TEXT" == "PARSE_FAIL"* ]] || [ -z "$TEXT" ]; then
echo -e "RED❌ 无返回内容NC"
ISSUES+=("MODEL_ID: 无返回内容")
else
echo -e "GREEN✅ 返回: TEXTNC"
OK_COUNT=$((OK_COUNT + 1))
fi
done
echo ""
done <<< "$PROVIDERS"
rm -f "$TMPPARSE" "$TMPPARSE_OAI"
echo "=========================================="
echo "📊 校验总结"
echo "=========================================="
echo "可用模型: OK_COUNT/TOTAL"
echo ""
if [ #ISSUES[@] -eq 0 ]; then
echo -e "GREEN🎉 所有模型配置正常!NC"
else
echo -e "RED⚠️ 发现 #ISSUES[@] 个问题:NC"
for issue in "ISSUES[@]"; do
echo " • issue"
done
echo ""
echo "💡 修复建议:"
echo " - apiKey 无效 → 更新配置中的 apiKey"
echo " - HTTP 500 → 检查账户余额或联系服务商"
echo " - 连接超时 → 检查网络或 baseUrl"
echo " - 无返回内容 → 检查 model id 是否正确、max_tokens 是否充足"
fi
Control Xiaomi Mi Home (米家) smart devices via Xiaomi Cloud API. Use when: user wants to control smart home devices (lights, AC, heater, bath heater, switches...
---
name: mijia-control
description: |
Control Xiaomi Mi Home (米家) smart devices via Xiaomi Cloud API.
Use when: user wants to control smart home devices (lights, AC, heater, bath heater, switches, etc.),
check device status, or create automation scenes.
Triggers: "开灯", "关灯", "开空调", "关空调", "开浴霸", "我要洗澡", "设备状态",
"创建场景", "自动化", "米家", "smart home", "turn on/off", any home device control request.
NOT for: non-Xiaomi devices, HomeKit-only devices.
---
# Mi Home (米家) Control
Control Xiaomi smart home devices and create automations via cloud API.
## Setup
**Prerequisites:** `pip3 install micloud`
**Credentials:** `~/.mijia_creds.json` (chmod 600)
```json
{"userId":"...", "serviceToken":"...", "ssecurity":"...", "cUserId":"..."}
```
If login requires 2FA (device trust verification), obtain credentials via browser:
1. Open `https://account.xiaomi.com/pass/serviceLogin?sid=xiaomiio&_json=true` in browser with existing cookies
2. POST to `serviceLoginAuth2` with hash password to get `ssecurity` + `location`
3. Follow `location` URL to get `serviceToken` from cookies
## Device Control
### List devices
```bash
python3 scripts/mijia.py devices
```
### Read property
```bash
python3 scripts/mijia.py get <did> <siid> <piid>
```
### Set property
```bash
python3 scripts/mijia.py set <did> <siid> <piid> <value>
# value: true/false for bool, integer for uint8, etc.
```
**Note:** Mesh devices (BLE switches) return code:1 on set — this is normal (async confirmation). The command still executes. Verify by reading the property after a 1-2 second delay.
### Batch control
Create a JSON file with commands:
```json
[
{"did": "946635824", "siid": 2, "piid": 1, "value": true},
{"did": "946633803", "siid": 2, "piid": 1, "value": false}
]
```
```bash
python3 scripts/mijia.py batch /tmp/commands.json
```
## MIoT Spec Reference
Each device has services (siid) → properties (piid). Look up specs at:
`https://home.miot-spec.com/spec/<model>`
Common patterns:
- **Switch/Light:** siid:2 piid:1 = on/off (bool)
- **3-way switch:** siid:2/3/4 piid:1 = channel 1/2/3 (bool)
- **AC:** siid:2 piid:1 = on/off, piid:2 = mode, piid:3 = target temp
- **Bath heater:** siid:2 = light, siid:3 = PTC heater, siid:10 = environment (temp sensor)
- **Water heater:** siid:2 piid:1 = on/off, piid:2 = target temp, piid:3 = current temp
## Create Automation Scenes
```bash
python3 scripts/mijia.py scene_create scene.json
```
Scene JSON format (see `references/scene_template.json`):
```json
{
"name": "回家自动开灯",
"identify": "homelight_1234",
"st_id": 8,
"setting": {"enable": 1, "enable_push": 0},
"trigger": {
"key": "event.<did>.<siid>.<eiid>",
"did": "<trigger_device_did>",
"model": "<model>",
"extra": "{\"siid\":2,\"eiid\":1}"
},
"action_list": [
{
"did": "<target_device_did>",
"model": "<model>",
"extra": "{\"props\":[{\"siid\":2,\"piid\":1,\"value\":true}]}",
"type": "device_ctrl"
}
]
}
```
Key fields:
- `trigger.key`: format `event.<did>.<siid>.<eiid>` for device events
- `launch.attr_filter`: optional conditions (e.g., only if light is off)
- `action_list[].extra`: JSON string with `props` array
- `st_id`: 8 = user automation
- `home_id` and `uid` are auto-filled if omitted
## Predefined Scenes
### 🚿 Bath Mode (洗澡模式)
When user says "我要洗澡":
1. Read bathroom temp: `get 927361805 10 1`
2. If temp < 28°C → turn on bath heater: `set 927361805 3 1 1` (暖风), target 30°C
3. If temp ≥ 28°C → skip heater
4. Turn on water heater: `set 965879649 2 1 1`, set 43°C: `set 965879649 2 2 43`
5. Turn on bathroom light: `set 946635824 2 1 true`
### 🐟 Auto Fish Feeding (出门自动喂鱼)
Feeds fish once per day when BOSS leaves home.
**Architecture:**
1. **OpenClaw cron** (every 5min): polls door lock event log via `/user/get_user_device_data`
2. If today has an auto-lock event (action=1) AND not fed yet → feed fish → record state
3. After feeding, all subsequent checks skip immediately (daily dedup)
4. **State file** `~/.fish_feed_state.json`: tracks last feed date
**Components:**
- Door lock events: 智能门锁2 Pro (did: 1175215651), event key `2.1020`, action=1 = auto-lock (leaving home)
- Fish feeder: 智能鱼缸 (did: 2026943875), action siid:2 aiid:1, in:[1] (1 portion)
- ⚠️ Action 参数格式: `"in": [1]` 而不是 `[{"piid":5,"value":1}]`
- Script: `scripts/fish_auto_feed.py`
**Manual commands:**
```bash
python3 scripts/fish_auto_feed.py # Normal run
python3 scripts/fish_auto_feed.py --dry-run # Check without feeding
python3 scripts/fish_auto_feed.py --force # Feed regardless
python3 scripts/fish_auto_feed.py --status # Show state
```
**Trigger logic:**
- Polls lock event log for auto-lock (action=1) today → feed fish
- Once fed, skips all further checks until next day
- No米家自动化 needed, pure OpenClaw polling
## 小爱音箱 TTS
Make the XiaoAI speaker say something aloud.
```bash
python3 scripts/mijia.py tts <did> "<text>"
# Example: python3 scripts/mijia.py tts 100783118 "你好"
```
Flow: unmute → play-text → wait → pause playback → re-mute.
This prevents the speaker from auto-playing music after TTS.
**Important:**
- Speaker may be muted (default state at night) — script handles unmute/re-mute automatically
- After TTS, always pause playback to prevent auto-music
- Cloud API cannot control screen UI — returning to clock display requires physical tap or voice command
## Token Refresh
Tokens expire periodically. If API returns decode errors or auth failures:
1. Re-login via browser (same flow as initial setup)
2. Update `~/.mijia_creds.json` with new `serviceToken` and `ssecurity`
The `ssecurity` changes each login. The `serviceToken` is session-bound.
FILE:references/devices.json
{
"_doc": "BOSS home device registry. Lookup by Chinese name to find did/siid/piid.",
"devices": {
"厕所灯": {"did": "946635824", "model": "xiaomi.switch.2wpro3", "type": "3-way-switch"},
"客厅灯": {"did": "946633803", "model": "xiaomi.switch.2wpro3", "type": "3-way-switch"},
"入户门开关": {"did": "946796488", "model": "xiaomi.switch.2wpro3", "type": "3-way-switch"},
"客厅中间开关": {"did": "946360216", "model": "xiaomi.switch.2wpro3", "type": "3-way-switch"},
"次卧门口": {"did": "946625036", "model": "xiaomi.switch.2wpro3", "type": "3-way-switch"},
"主卧左开关": {"did": "946293001", "model": "xiaomi.switch.2pro3", "type": "3-way-switch"},
"主卧右开关": {"did": "946658849", "model": "xiaomi.switch.2pro3", "type": "3-way-switch"},
"小米智能开关Pro3": {"did": "946271120", "model": "xiaomi.switch.2pro3", "type": "3-way-switch"},
"家政间灯": {"did": "1034720433", "model": "zimi.switch.dhkg01", "type": "switch"},
"浴霸": {"did": "927361805", "model": "xiaomi.bhf_light.s1", "type": "bath-heater",
"notes": "siid:2=照明, siid:3=PTC暖风(piid:1=mode 0关/1暖风/2干燥/3吹风/4换气, piid:5=温度档), siid:10 piid:1=浴室温度"},
"热水器": {"did": "965879649", "model": "xiaomi.waterheater.ymm6", "type": "water-heater",
"notes": "siid:2 piid:1=开关(uint8 0/1), piid:2=目标温度, piid:3=当前水温"},
"主卧空调": {"did": "871747012", "model": "xiaomi.airc.arf1r1", "type": "ac"},
"客厅空调": {"did": "871759920", "model": "xiaomi.airc.arf1r1", "type": "ac"},
"次卧空调": {"did": "871783853", "model": "xiaomi.airc.arf1r1", "type": "ac"},
"客厅景观灯": {"did": "2042397495", "model": "lemesh.light.wy0d02", "type": "light"},
"小米电视EA": {"did": "613985952", "model": "xiaomi.tv.eaffh1", "type": "tv"},
"小爱触屏音箱": {"did": "100783118", "model": "xiaomi.wifispeaker.l04m", "type": "speaker"},
"智能门锁2 Pro": {"did": "1175215651", "model": "loock.lock.t3pul", "type": "lock"},
"智能鱼缸": {"did": "2026943875", "model": "hfjh.fishbowl.m100", "type": "fishbowl",
"notes": "siid:2=鱼缸(piid:1=开关, piid:2=水泵, piid:4=水温, piid:5=喂食量1-3), action siid:2 aiid:1=喂食(in:[份数], 不是{piid:5,value:N}格式)"},
"净烟机P1": {"did": "917614267", "model": "cykj.hood.jyj22", "type": "hood"},
"路由器": {"did": "miwifi.4405b803-d717-7957-d8a3-2807e4477844", "model": "xiaomi.router.ra72", "type": "router"},
"燃气热水器P10": {"did": "965879649", "model": "xiaomi.waterheater.ymm6", "type": "water-heater"}
},
"home_id": 159001331072,
"scenes": {
"洗澡模式": {
"temp_threshold": 28,
"bath_heater_target": 30,
"water_heater_target": 43
},
"出门自动喂鱼": {
"method": "OpenClaw cron polling lock events",
"trigger": "门锁自动锁门(action=1)",
"cron": "fish-auto-feed (every 5min)",
"state_file": "~/.fish_feed_state.json"
}
}
}
FILE:references/scene_template.json
{
"_doc": "Mi Home automation scene template. Copy and modify.",
"name": "场景名称",
"identify": "unique_id_timestamp",
"st_id": 8,
"setting": {"enable": 1, "enable_push": 0},
"trigger": {
"_doc": "Trigger: device event. key format: event.<did>.<siid>.<eiid>",
"key": "event.<did>.<siid>.<eiid>",
"did": "<device_did>",
"model": "<device_model>",
"extra": "{\"siid\":2,\"eiid\":1}"
},
"launch": {
"_doc": "Optional conditions (only execute if conditions met)",
"attr_filter": {
"conditions": [
{
"did": "<device_did>",
"model": "<device_model>",
"extra": "{\"siid\":2,\"piid\":1,\"value\":false,\"operator\":\"=\"}"
}
],
"logic": "and"
}
},
"action_list": [
{
"_doc": "Action: control a device",
"did": "<device_did>",
"model": "<device_model>",
"extra": "{\"props\":[{\"siid\":2,\"piid\":1,\"value\":true}]}",
"type": "device_ctrl"
}
],
"home_id": 0,
"uid": 0
}
FILE:scripts/fish_auto_feed.py
#!/usr/bin/env python3
"""
Auto Fish Feeder v2 - polls door lock events and feeds fish once per day.
Flow:
1. OpenClaw cron (every 5min) runs this script
2. Script checks door lock event log for auto-lock events (action=1) today
3. If found and not fed today → feed fish → record state → done
4. If already fed today → skip immediately
Usage:
python3 fish_auto_feed.py # Normal run
python3 fish_auto_feed.py --dry-run # Check without feeding
python3 fish_auto_feed.py --force # Feed regardless of state
python3 fish_auto_feed.py --status # Show current state
"""
import sys, os, json, time, datetime
from pathlib import Path
STATE_FILE = Path.home() / ".fish_feed_state.json"
MIJIA_SCRIPT = Path(__file__).parent.parent / "skills" / "mijia-control" / "scripts" / "mijia.py"
# Config
LOCK_DID = "1175215651" # 智能门锁2 Pro
LOCK_EVENT_KEY = "2.1020" # Lock Event
FISH_DID = "2026943875" # 智能鱼缸
FISH_FEED_SIID = 2
FISH_FEED_AIID = 1
FEED_AMOUNT = 1 # 1-3 portions
def load_state():
if STATE_FILE.exists():
return json.loads(STATE_FILE.read_text())
return {}
def save_state(state):
STATE_FILE.write_text(json.dumps(state, indent=2))
def get_cloud():
"""Import and init micloud from mijia.py."""
import importlib.util
spec = importlib.util.spec_from_file_location("mijia_mod", str(MIJIA_SCRIPT))
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod.get_cloud()
def check_lock_events_today(mc):
"""Check if there's an auto-lock event (leaving home) today."""
now_ts = int(time.time())
now_dt = datetime.datetime.now()
# Cycle starts at 06:00 today, or 06:00 yesterday if before 06:00
if now_dt.hour < 6:
cycle_start = (now_dt - datetime.timedelta(days=1)).replace(hour=6, minute=0, second=0, microsecond=0)
else:
cycle_start = now_dt.replace(hour=6, minute=0, second=0, microsecond=0)
ts_start = int(cycle_start.timestamp())
params = {"data": json.dumps({
"did": LOCK_DID,
"key": LOCK_EVENT_KEY,
"type": "event",
"time_start": ts_start,
"time_end": now_ts,
"limit": 20
})}
r = mc.request_country("/user/get_user_device_data", "cn", params)
data = json.loads(r if isinstance(r, str) else r.decode())
events = data.get("result", [])
for evt in events:
try:
props = json.loads(evt["value"])
prop_map = {p["piid"]: p["value"] for p in props}
action = prop_map.get(3) # piid:3 = Lock Action
# action=1 means Lock (auto-lock after leaving)
if action == 1:
evt_time = datetime.datetime.fromtimestamp(evt["time"])
print(f"[FOUND] Auto-lock event at {evt_time.strftime('%H:%M:%S')}")
return True
except:
continue
return False
def feed_fish(mc):
"""Trigger fish feeder via miotspec action."""
params = {"data": json.dumps({"params": {
"did": FISH_DID,
"siid": FISH_FEED_SIID,
"aiid": FISH_FEED_AIID,
"in": [FEED_AMOUNT]
}})}
r = mc.request_country("/miotspec/action", "cn", params)
data = json.loads(r if isinstance(r, str) else r.decode())
code = data.get("result", {}).get("code", -1)
print(f"[INFO] Fish feed API response code: {code}")
return code in (0, 1)
def main():
dry_run = "--dry-run" in sys.argv
force = "--force" in sys.argv
status = "--status" in sys.argv
# "Today" runs from 06:00 to next day 06:00
now = datetime.datetime.now()
if now.hour < 6:
feed_day = (now.date() - datetime.timedelta(days=1)).isoformat()
else:
feed_day = now.date().isoformat()
state = load_state()
if status:
print(json.dumps(state, indent=2, ensure_ascii=False))
return
# Already fed this cycle? Skip entirely.
if state.get("last_feed_date") == feed_day and not force:
print(f"[OK] Already fed this cycle ({feed_day}). Done.")
return
mc = get_cloud()
# Check for auto-lock event today
has_left = check_lock_events_today(mc)
if not has_left and not force:
print(f"[OK] No auto-lock event today. BOSS hasn't left home yet.")
return
print(f"[TRIGGER] BOSS has left home. Feeding fish...")
if dry_run:
print("[DRY-RUN] Would feed fish now. Skipping.")
return
success = feed_fish(mc)
# Record state regardless of success — only attempt once per day
state["last_feed_date"] = feed_day
state["last_feed_time"] = datetime.datetime.now().isoformat()
state["last_feed_success"] = success
state["total_feeds"] = state.get("total_feeds", 0) + (1 if success else 0)
save_state(state)
if success:
print(f"[DONE] Fed fish. Total feeds: {state['total_feeds']}")
else:
print(f"[WARN] Feeding may have failed. Will NOT retry today.")
if __name__ == "__main__":
main()
FILE:scripts/mijia.py
#!/usr/bin/env python3
"""
Mi Home (米家) Cloud Controller
Controls Xiaomi smart home devices via Xiaomi Cloud API.
Usage:
python3 mijia.py login # Login and save credentials
python3 mijia.py devices # List all devices
python3 mijia.py status <did> # Read device properties
python3 mijia.py set <did> <siid> <piid> <val> # Set a device property
python3 mijia.py get <did> <siid> <piid> # Get a device property
python3 mijia.py scene_create <json_file> # Create automation scene
python3 mijia.py scene_list # List automation scenes
python3 mijia.py batch <json_file> # Execute batch commands
Credentials are stored in ~/.mijia_creds.json (auto-refreshed via browser cookies).
"""
import sys, os, json, hashlib, time, argparse
from pathlib import Path
# Ensure micloud is available
try:
from micloud import MiCloud
from micloud.miutils import get_session
except ImportError:
print("ERROR: micloud not installed. Run: pip3 install micloud", file=sys.stderr)
sys.exit(1)
CREDS_PATH = Path.home() / ".mijia_creds.json"
DEVICES_CACHE = Path.home() / ".mijia_devices.json"
def load_creds():
if not CREDS_PATH.exists():
print(f"ERROR: No credentials found at {CREDS_PATH}", file=sys.stderr)
print("Run 'python3 mijia.py login' first, or create the file manually.", file=sys.stderr)
sys.exit(1)
return json.loads(CREDS_PATH.read_text())
def save_creds(creds):
CREDS_PATH.write_text(json.dumps(creds, indent=2))
CREDS_PATH.chmod(0o600)
def get_cloud():
creds = load_creds()
mc = MiCloud.__new__(MiCloud)
mc.user_id = int(creds["userId"])
mc.service_token = creds["serviceToken"]
mc.ssecurity = creds["ssecurity"]
mc.cuser_id = creds.get("cUserId", "")
mc.session = None
mc.locale = "zh_CN"
mc.timezone = "GMT+08:00"
mc.default_server = "cn"
mc.failed_logins = 0
return mc
def cmd_login(args):
"""Interactive login - saves credentials."""
username = args.username or input("Xiaomi username (phone): ")
password = args.password or input("Xiaomi password: ")
mc = MiCloud(username, password)
try:
mc.login()
creds = {
"userId": str(mc.user_id),
"serviceToken": mc.service_token,
"ssecurity": mc.ssecurity,
"cUserId": mc.cuser_id or ""
}
save_creds(creds)
print(f"Login successful! Credentials saved to {CREDS_PATH}")
except Exception as e:
print(f"Login failed: {e}", file=sys.stderr)
print("If 2FA is required, obtain credentials via browser and save manually.", file=sys.stderr)
sys.exit(1)
def cmd_devices(args):
"""List all devices."""
mc = get_cloud()
devices = mc.get_devices(country="cn")
if not devices:
print("No devices found or API error.")
return
# Cache devices
DEVICES_CACHE.write_text(json.dumps(devices, ensure_ascii=False, indent=2))
print(f"Found {len(devices)} devices:\n")
for d in devices:
online = "🟢" if d.get("isOnline") else "🔴"
print(f"{online} {d.get('name')} | model: {d.get('model')} | did: {d.get('did')} | ip: {d.get('localip','N/A')}")
def cmd_get(args):
"""Get a device property."""
mc = get_cloud()
params = {"data": json.dumps({"params": [
{"did": args.did, "siid": int(args.siid), "piid": int(args.piid)}
]})}
r = mc.request_country("/miotspec/prop/get", "cn", params)
data = json.loads(r if isinstance(r, str) else r.decode())
result = data.get("result", [{}])[0]
if result.get("code") == 0:
print(json.dumps({"value": result["value"]}, ensure_ascii=False))
else:
print(json.dumps({"error": result.get("code")}, ensure_ascii=False))
def cmd_set(args):
"""Set a device property."""
mc = get_cloud()
# Auto-detect value type
val = args.value
if val.lower() == "true": val = True
elif val.lower() == "false": val = False
else:
try: val = int(val)
except:
try: val = float(val)
except: pass
params = {"data": json.dumps({"params": [
{"did": args.did, "siid": int(args.siid), "piid": int(args.piid), "value": val}
]})}
r = mc.request_country("/miotspec/prop/set", "cn", params)
data = json.loads(r if isinstance(r, str) else r.decode())
result = data.get("result", [{}])[0]
code = result.get("code", -1)
# code:0 = confirmed success, code:1 = sent (mesh async), both are OK
if code in (0, 1):
print(json.dumps({"ok": True, "code": code}, ensure_ascii=False))
else:
print(json.dumps({"ok": False, "code": code}, ensure_ascii=False), file=sys.stderr)
sys.exit(1)
def cmd_status(args):
"""Read common properties of a device."""
mc = get_cloud()
# Read siid 2-10, piid 1-5
props = []
for siid in range(2, 11):
for piid in range(1, 6):
props.append({"did": args.did, "siid": siid, "piid": piid})
params = {"data": json.dumps({"params": props})}
r = mc.request_country("/miotspec/prop/get", "cn", params)
data = json.loads(r if isinstance(r, str) else r.decode())
results = []
for p in data.get("result", []):
if p.get("code") == 0:
results.append({"siid": p["siid"], "piid": p["piid"], "value": p["value"]})
print(json.dumps(results, ensure_ascii=False, indent=2))
def cmd_batch(args):
"""Execute batch set commands from JSON file."""
mc = get_cloud()
commands = json.loads(Path(args.json_file).read_text())
params_list = []
for cmd in commands:
params_list.append({
"did": cmd["did"],
"siid": cmd["siid"],
"piid": cmd["piid"],
"value": cmd["value"]
})
params = {"data": json.dumps({"params": params_list})}
r = mc.request_country("/miotspec/prop/set", "cn", params)
data = json.loads(r if isinstance(r, str) else r.decode())
for result in data.get("result", []):
code = result.get("code", -1)
ok = "✅" if code in (0, 1) else "❌"
print(f"{ok} did:{result['did']} siid:{result['siid']} piid:{result['piid']} code:{code}")
def cmd_tts(args):
"""Make a XiaoAI speaker say text aloud."""
import time as _time
mc = get_cloud()
did = args.did
text = args.text
def _action(siid, aiid, ins):
params = {"data": json.dumps({"params": {"did": did, "siid": siid, "aiid": aiid, "in": ins}})}
r = mc.request_country("/miotspec/action", "cn", params)
return json.loads(r if isinstance(r, str) else r.decode())
def _set(props):
params = {"data": json.dumps({"params": [{"did": did, **p} for p in props]})}
mc.request_country("/miotspec/prop/set", "cn", params)
# 1. Unmute + ensure volume
_set([{"siid": 2, "piid": 2, "value": False}, {"siid": 2, "piid": 1, "value": int(args.volume)}])
_time.sleep(0.5)
# 2. Play text (siid:5 aiid:1, in: piid:1=text)
result = _action(5, 1, [{"piid": 1, "value": text}])
code = result.get("result", {}).get("code", -1)
print(json.dumps({"ok": code == 0, "code": code}, ensure_ascii=False))
# 3. Wait for speech to finish (~0.15s per char + 2s buffer)
wait = max(2, len(text) * 0.15 + 2)
_time.sleep(wait)
# 4. Pause playback to prevent auto-music
_action(3, 2, [])
# 5. Re-mute if requested
if args.remute:
_set([{"siid": 2, "piid": 2, "value": True}])
def cmd_scene_create(args):
"""Create an automation scene from JSON definition."""
mc = get_cloud()
scene = json.loads(Path(args.json_file).read_text())
# Ensure required fields
if "home_id" not in scene:
scene["home_id"] = _get_home_id(mc)
if "uid" not in scene:
scene["uid"] = mc.user_id
if "st_id" not in scene:
scene["st_id"] = 8
if "setting" not in scene:
scene["setting"] = {"enable": 1, "enable_push": 0}
if "identify" not in scene:
scene["identify"] = f"scene_{int(time.time())}"
params = {"data": json.dumps(scene)}
r = mc.request_country("/scene/edit", "cn", params)
if isinstance(r, bytes):
r = r.decode()
data = json.loads(r) if isinstance(r, str) else r
print(json.dumps(data, ensure_ascii=False, indent=2))
def cmd_scene_list(args):
"""List existing scenes."""
mc = get_cloud()
home_id = _get_home_id(mc)
params = {"data": json.dumps({"home_id": home_id, "req_type": 1})}
r = mc.request_country("/appgateway/miot/appsceneservice/AppSceneService/GetSceneList", "cn", params)
data = json.loads(r if isinstance(r, str) else r.decode())
if data.get("code") == 0:
scenes = data.get("result", {}).get("scene_info_list", [])
print(f"Found {len(scenes)} scenes:")
for s in scenes:
print(f" - {s.get('name','')} | id:{s.get('scene_id','')}")
else:
print(f"Error: {data}")
def _get_home_id(mc):
"""Get the first home ID."""
params = {"data": json.dumps({"fg": True, "fetch_share": True, "limit": 10, "app_ver": 7})}
r = mc.request_country("/v2/homeroom/gethome", "cn", params)
data = json.loads(r if isinstance(r, str) else r.decode())
homes = data.get("result", {}).get("homelist", [])
return homes[0]["id"] if homes else 0
def main():
parser = argparse.ArgumentParser(description="Mi Home Cloud Controller")
sub = parser.add_subparsers(dest="command")
p_login = sub.add_parser("login", help="Login to Xiaomi Cloud")
p_login.add_argument("--username", "-u")
p_login.add_argument("--password", "-p")
sub.add_parser("devices", help="List all devices")
p_get = sub.add_parser("get", help="Get device property")
p_get.add_argument("did")
p_get.add_argument("siid")
p_get.add_argument("piid")
p_set = sub.add_parser("set", help="Set device property")
p_set.add_argument("did")
p_set.add_argument("siid")
p_set.add_argument("piid")
p_set.add_argument("value")
p_status = sub.add_parser("status", help="Read device status")
p_status.add_argument("did")
p_batch = sub.add_parser("batch", help="Batch set from JSON")
p_batch.add_argument("json_file")
p_sc = sub.add_parser("scene_create", help="Create automation scene")
p_sc.add_argument("json_file")
p_tts = sub.add_parser("tts", help="Make speaker say text")
p_tts.add_argument("did")
p_tts.add_argument("text")
p_tts.add_argument("--volume", type=int, default=50)
p_tts.add_argument("--no-remute", dest="remute", action="store_false", default=True)
sub.add_parser("scene_list", help="List scenes")
args = parser.parse_args()
commands = {
"login": cmd_login, "devices": cmd_devices, "get": cmd_get,
"set": cmd_set, "status": cmd_status, "batch": cmd_batch, "tts": cmd_tts,
"scene_create": cmd_scene_create, "scene_list": cmd_scene_list
}
if args.command in commands:
commands[args.command](args)
else:
parser.print_help()
if __name__ == "__main__":
main()
Send images directly in Feishu chat as native image messages (not file attachments). Use when: need to send a generated image, chart, screenshot, or any loca...
--- name: feishu-send-image description: | Send images directly in Feishu chat as native image messages (not file attachments). Use when: need to send a generated image, chart, screenshot, or any local image file to a Feishu user or group chat. Triggers: "发图", "send image", "发送图片", sending charts/graphs/screenshots via Feishu. Note: OpenClaw's built-in message tool (filePath/media/buffer) does NOT work for Feishu image sending — it only produces file attachments. This skill bypasses that limitation by calling Feishu Bot API directly. --- # Feishu Send Image Send local image files as native Feishu image messages via the Feishu Bot API. ## Why This Skill Exists OpenClaw's `message` tool with `filePath`, `media`, or `buffer` parameters only sends file attachments in Feishu, not inline images. This skill calls the Feishu API directly to send proper image messages. ## Quick Usage Run the script: ```bash bash scripts/feishu_send_image.sh <image_path> <receive_id> <app_id> <app_secret> [receive_id_type] ``` ### Arguments | Arg | Description | |-----|-------------| | `image_path` | Local path to image (png/jpg/gif/webp) | | `receive_id` | Feishu `open_id` (user) or `chat_id` (group) | | `app_id` | Feishu app ID from `~/.openclaw/openclaw.json` → `feishu.accounts.default.appId` | | `app_secret` | Feishu app secret from `~/.openclaw/openclaw.json` → `feishu.accounts.default.appSecret` | | `receive_id_type` | `open_id` (default) or `chat_id` | ### Example ```bash bash scripts/feishu_send_image.sh /tmp/chart.png \ ou_38470740452f6083ce189b7ddec722f8 \ cli_a92c368412f9dcb1 \ 7uM7aLqeqYqm0Fsy0IP5QhOyTBSwxlfT ``` ## Getting Credentials Read `~/.openclaw/openclaw.json` and extract: - `channels.feishu.accounts.default.appId` - `channels.feishu.accounts.default.appSecret` The receiver's `open_id` comes from inbound message metadata (`sender_id`). ## How It Works 1. **Get token** — `POST /auth/v3/tenant_access_token/internal` with appId/appSecret 2. **Upload image** — `POST /im/v1/images` with `image_type=message`, returns `image_key` 3. **Send message** — `POST /im/v1/messages` with `msg_type=image` and the `image_key` ## Output On success: `OK: image_key=<key> message_id=<id>` On failure: prints error and exits with code 1. FILE:README.md # feishu-send-image 在飞书聊天中发送原生图片消息的 OpenClaw 技能。 ## 解决什么问题 OpenClaw 内置的 `message` 工具在飞书中发送图片时,无论使用 `filePath`、`media` 还是 `buffer` 参数,都只会生成文件附件,而非可直接预览的图片消息。 本技能通过直接调用飞书 Bot API,实现真正的图片消息发送。 ## 使用场景 - 发送 AI 生成的图表(股票走势、数据分析图等) - 发送截图、处理后的图片 - 任何需要在飞书对话中直接展示图片的场景 ## 工作原理 三步完成图片发送: 1. 获取飞书 `tenant_access_token` 2. 上传图片获取 `image_key` 3. 发送 `image` 类型消息 ## 使用方法 ```bash bash scripts/feishu_send_image.sh <图片路径> <接收者ID> <app_id> <app_secret> [id类型] ``` ### 参数说明 | 参数 | 说明 | |------|------| | `图片路径` | 本地图片文件路径,支持 png/jpg/gif/webp | | `接收者ID` | 飞书用户的 `open_id` 或群聊的 `chat_id` | | `app_id` | 飞书应用 App ID | | `app_secret` | 飞书应用 App Secret | | `id类型` | 可选,`open_id`(默认)或 `chat_id` | ### 示例 ```bash # 发送图片给用户 bash scripts/feishu_send_image.sh /tmp/chart.png ou_xxxxx cli_xxxxx secret_xxxxx # 发送图片到群聊 bash scripts/feishu_send_image.sh /tmp/chart.png oc_xxxxx cli_xxxxx secret_xxxxx chat_id ``` ## 凭证获取 App ID 和 App Secret 可从 OpenClaw 配置文件中获取: ``` ~/.openclaw/openclaw.json → channels.feishu.accounts.default.appId / appSecret ``` 接收者 `open_id` 来自飞书消息的 `sender_id` 字段。 ## 前置条件 - 飞书自建应用已创建并启用机器人能力 - 应用已获取 `im:message:send_as_bot`(发送消息)和 `im:resource`(上传图片)权限 - 系统已安装 `curl` 和 `python3` ## 安装 ```bash openclaw skill install feishu-send-image.skill ``` 或手动将 `feishu-send-image/` 目录放入 `~/.openclaw/workspace/skills/` 下。 FILE:scripts/feishu_send_image.sh #!/bin/bash # Send an image to a Feishu user/chat via Feishu Bot API # Usage: feishu_send_image.sh <image_path> <receive_id> <app_id> <app_secret> [receive_id_type] # # Arguments: # image_path - Local path to the image file (png/jpg/gif/webp) # receive_id - Feishu open_id (user) or chat_id (group) # app_id - Feishu app ID # app_secret - Feishu app secret # receive_id_type - "open_id" (default) or "chat_id" set -euo pipefail IMAGE_PATH="?Usage: feishu_send_image.sh <image_path> <receive_id> <app_id> <app_secret> [receive_id_type]" RECEIVE_ID="?Missing receive_id" APP_ID="?Missing app_id" APP_SECRET="?Missing app_secret" RECEIVE_ID_TYPE="-open_id" # Step 1: Get tenant_access_token TOKEN_RESP=$(curl -s -X POST 'https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal' \ -H 'Content-Type: application/json' \ -d "{\"app_id\":\"$APP_ID\",\"app_secret\":\"$APP_SECRET\"}") TOKEN=$(echo "$TOKEN_RESP" | python3 -c "import sys,json; print(json.load(sys.stdin)['tenant_access_token'])" 2>/dev/null) if [ -z "$TOKEN" ]; then echo "ERROR: Failed to get access token" >&2 echo "$TOKEN_RESP" >&2 exit 1 fi # Step 2: Upload image to get image_key UPLOAD_RESP=$(curl -s -X POST 'https://open.feishu.cn/open-apis/im/v1/images' \ -H "Authorization: Bearer $TOKEN" \ -F 'image_type=message' \ -F "image=@$IMAGE_PATH") IMAGE_KEY=$(echo "$UPLOAD_RESP" | python3 -c "import sys,json; print(json.load(sys.stdin)['data']['image_key'])" 2>/dev/null) if [ -z "$IMAGE_KEY" ]; then echo "ERROR: Failed to upload image" >&2 echo "$UPLOAD_RESP" >&2 exit 1 fi # Step 3: Send image message SEND_RESP=$(curl -s -X POST "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=$RECEIVE_ID_TYPE" \ -H "Authorization: Bearer $TOKEN" \ -H "Content-Type: application/json" \ -d "{\"receive_id\":\"$RECEIVE_ID\",\"msg_type\":\"image\",\"content\":\"{\\\"image_key\\\":\\\"$IMAGE_KEY\\\"}\"}") CODE=$(echo "$SEND_RESP" | python3 -c "import sys,json; print(json.load(sys.stdin).get('code',1))" 2>/dev/null) if [ "$CODE" = "0" ]; then MSG_ID=$(echo "$SEND_RESP" | python3 -c "import sys,json; print(json.load(sys.stdin)['data']['message_id'])" 2>/dev/null) echo "OK: image_key=$IMAGE_KEY message_id=$MSG_ID" else echo "ERROR: Failed to send message" >&2 echo "$SEND_RESP" >&2 exit 1 fi