agent-skills
/
excel-toolkit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

refactor: remove embedded python_auth_runtime, use shared auth-runtime-py 

Eliminates the third copy of auth logic. Dependency now points to the
shared auth-runtime-py package. Removes AUTH_RUNTIME.md and CLIENT_KEY
from .env.example.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
ywkj 2026-03-19 08:31:55 +08:00
parent 585cc659b8
commit 325834eff4
9 changed files with 828 additions and 1831 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
.env.example
View File

@ -11,18 +11,4 @@ GEMINI_API_KEY=your-gemini-api-key-here
# Gemini 模型名称（可选，默认：gemini-2.0-flash-lite）
# GEMINI_MODEL=gemini-2.0-flash-lite
# -----------------------------------------------------------------------------
# OpenClaw Auth Runtime 配置（用于 API 访问）
# 从 ~/clawd/skills/_shared 获取 CLIENT_KEY
# -----------------------------------------------------------------------------
# 认证基础 URL（默认：https://api-gw-test.yuanwei-lnc.com）
# AUTH_BASE=https://api-gw-test.yuanwei-lnc.com
# 客户端密钥（必需）- 从 ~/clawd/skills/_shared 获取
# CLIENT_KEY=your-client-key-here
# 缓存目录（默认：/tmp/skill-auth-cache）
# AUTH_CACHE_DIR=/tmp/skill-auth-cache
# 最小令牌 TTL 秒数（默认：60）
# AUTH_MIN_TTL_SEC=60
# Auth is handled automatically via ~/.openclaw/.env

215
AUTH_RUNTIME.md
View File

@ -1,215 +0,0 @@
# OpenClaw Auth Runtime Python 实现
基于 `~/clawd/skills/_shared/auth-runtime` 的 TypeScript 实现，提供 Python 版本的客户端密钥认证。
## 功能特性
- ✅ 使用 CLIENT_KEY 获取访问令牌
- ✅ 支持令牌缓存（可配置 TTL）
- ✅ 自动刷新过期令牌
- ✅ 401/403 自动重试
- ✅ 从 .env 文件自动加载配置
## 快速开始
### 1. 配置环境变量
在 `.env` 文件中设置 CLIENT_KEY：
```bash
# 从 ~/clawd/skills/_shared 获取 CLIENT_KEY
CLIENT_KEY=your-client-key-here
```
### 2. 获取访问令牌
```python
from auth_runtime import create_env_config, get_access_token
# 创建配置（自动从环境变量加载）
config = create_env_config()
# 获取访问令牌（带缓存）
token = get_access_token(dry_run=False, config=config)
```
### 3. 发送 API 请求
```python
from auth_runtime import request_api_with_auto_refresh
# 发送请求（自动处理令牌刷新）
response = request_api_with_auto_refresh(
method="POST",
url="https://api-gw-test.yuanwei-lnc.com/your-api-endpoint",
dry_run=False,
config=config,
body={"key": "value"},
)
print(f"状态码：{response.status}")
print(f"响应体：{response.body}")
```
## API 参考
### 配置
#### `create_env_config() -> EnvConfig`
从环境变量创建配置：
| 环境变量 | 默认值 | 说明 |
|---------|--------|------|
| `AUTH_BASE` | `https://api-gw-test.yuanwei-lnc.com` | 认证基础 URL |
| `CLIENT_KEY` | **必需** | 客户端密钥 |
| `AUTH_CACHE_DIR` | `/tmp/skill-auth-cache` | 缓存目录 |
| `AUTH_MIN_TTL_SEC` | `60` | 最小令牌 TTL（秒） |
### 令牌管理
#### `get_access_token(dry_run, config) -> str`
获取访问令牌（带缓存）：
1. 检查缓存
2. 如果缓存有效，返回缓存令牌
3. 否则请求新令牌并缓存
#### `refresh_access_token(dry_run, config) -> str`
刷新访问令牌（绕过缓存）
### API 请求
#### `request_api(method, url, auth_token, body) -> ApiResponse`
发送 HTTP 请求
#### `request_api_with_auto_refresh(method, url, dry_run, config, body) -> ApiResponse`
发送 API 请求并自动刷新令牌：
1. 使用当前令牌发送请求
2. 如果是 401/403 错误，刷新令牌并重试一次
### 便捷函数
#### `load_client_key_from_env(env_path) -> str`
从 .env 文件加载 CLIENT_KEY
优先级：
1. 环境变量 `CLIENT_KEY`
2. .env 文件中的 `CLIENT_KEY`
3. 抛出异常
## 使用示例
### 示例 1: 简单的 API 调用
```python
from auth_runtime import create_env_config, request_api_with_auto_refresh
config = create_env_config()
response = request_api_with_auto_refresh(
method="POST",
url=f"{config.auth_base}/your-endpoint",
dry_run=False,
config=config,
body={"url": "https://example.com"},
)
if response.status == 200:
print("成功:", response.body)
else:
print("失败:", response.status, response.body)
```
### 示例 2: 结合 Excel 翻译使用
```python
from auth_runtime import create_env_config, request_api_with_auto_refresh
import json
config = create_env_config()
# 调用 1688 API 获取商品详情
response = request_api_with_auto_refresh(
method="POST",
url=f"{config.auth_base}/ecom/tasks/scrape",
dry_run=False,
config=config,
body={"url": "https://detail.1688.com/offer/123.html"},
)
if response.status == 200:
data = json.loads(response.body)
print("商品价格:", data["price"])
```
## 缓存机制
令牌缓存位置：`/tmp/skill-auth-cache/token_<hash>.json`
缓存文件内容：
```json
{
"access_token": "eyJhbGciOiJIUzI1NiIs...",
"expires_at": 1234567890.123,
"expires_in": 900
}
```
缓存策略：
- 令牌在过期前 `AUTH_MIN_TTL_SEC` 秒内视为无效
- 默认最小 TTL 为 60 秒
- 缓存文件自动创建和删除
## 错误处理
### 常见错误
| 错误 | 原因 | 解决方案 |
|------|------|---------|
| `CLIENT_KEY is required` | 未设置 CLIENT_KEY | 在 .env 文件或环境变量中设置 |
| `Auth session request failed` | 认证请求失败 | 检查 CLIENT_KEY 是否有效 |
| `Missing accessToken` | 认证响应缺少令牌 | 检查认证服务是否正常 |
### 自动重试
以下情况会自动刷新令牌并重试一次：
- HTTP 401 Unauthorized
- HTTP 403 Forbidden
- 响应体包含 "expired"、"unauthorized" 等标记
## 测试
运行测试脚本：
```bash
cd /Users/xiaolongxia/Documents/ai-build-app/skills/excel-toolkit
uv run python scripts/test_auth.py
```
## 与 TypeScript 版本的差异
| 特性 | TypeScript | Python |
|------|-----------|--------|
| HTTP 客户端 | 原生 fetch | requests |
| 缓存文件 | Bun/Node fs | pathlib |
| 环境变量 | process.env | os.getenv |
| 类型系统 | TypeScript | dataclass |
功能完全对等，可以互换使用。
## 获取 CLIENT_KEY
CLIENT_KEY 存储在 `~/clawd/skills/_shared` 目录中。
查看相关文档：
- `~/clawd/skills/_shared/AUTH_RUNTIME_IMPLEMENTATION.md`
- `~/clawd/skills/_shared/auth-runtime/README.md`
## 许可证
与 OpenClaw 项目保持一致。

4
pyproject.toml
View File

@ -7,9 +7,7 @@ dependencies = [
"pandas>=1.5.0",
"openpyxl>=3.0.0",
"google-generativeai>=0.8.0",
"requests>=2.28.0",
# 引用本地 auth-runtime 包
"clawd-auth-runtime-py @ file:///${PROJECT_ROOT}/python_auth_runtime",
"clawd-auth-runtime-py @ file:///${PROJECT_ROOT}/../auth-runtime-py",
]
[tool.uv]

237
python_auth_runtime/README.md
View File

@ -1,237 +0,0 @@
# @clawd/auth-runtime-py
Python 版本的 OpenClaw Auth Runtime。
## 全局配置
**所有 skill 共享同一份全局配置**，无需在每个 skill 中重复配置。
### 创建全局配置文件
```bash
# 复制模板
cp ~/.openclaw/.env.example ~/.openclaw/.env
# 编辑配置文件
vi ~/.openclaw/.env
```
### 配置文件内容
```bash
# ~/.openclaw/.env
# Auth Runtime 配置
CLIENT_KEY=sk_ae28fc4e.xxx
AUTH_BASE=https://api-gw-test.yuanwei-lnc.com
# Gemini API 配置（可选）
GEMINI_API_KEY=your-gemini-api-key
```
**注意**: `~/.openclaw/.env` 包含敏感信息，不要提交到 git。
---
## 安装
```bash
uv pip install /path/to/python_auth_runtime
```
---
## 使用方式
### 最简单的用法
```python
from python_auth_runtime import create_env_config, request_api_with_auto_refresh
# 自动从 ~/.openclaw/.env 加载配置
config = create_env_config()
# 调用 API
response = request_api_with_auto_refresh(
method="POST",
url=f"{config.auth_base}/ecom/tasks/scrape",
dry_run=False,
config=config,
body={"url": "https://detail.1688.com/offer/123.html"},
)
```
### 配置加载优先级
1. **环境变量**（最高优先级）
```bash
export CLIENT_KEY="override-key"
python script.py
```
2. **全局配置文件** `~/.openclaw/.env`
```bash
CLIENT_KEY=sk_ae28fc4e.xxx
```
3. **默认值**
- `AUTH_BASE`: `https://api-gw-test.yuanwei-lnc.com`
- `AUTH_CACHE_DIR`: `/tmp/skill-auth-cache`
- `AUTH_MIN_TTL_SEC`: `60`
---
## 完整示例
```python
#!/usr/bin/env python3
# your-skill/scripts/main.py
from python_auth_runtime import create_env_config, request_api_with_auto_refresh
import json
def main():
# 自动从 ~/.openclaw/.env 加载配置
config = create_env_config()
# 验证配置
if not config.client_key:
print("❌ CLIENT_KEY not found!")
print("Please create ~/.openclaw/.env with your CLIENT_KEY")
print("See: ~/.openclaw/.env.example")
return 1
# 调用 1688 API
response = request_api_with_auto_refresh(
method="POST",
url=f"{config.auth_base}/ecom/tasks/scrape",
dry_run=False,
config=config,
body={"url": "https://detail.1688.com/offer/123.html"},
)
if response.status == 200:
data = json.loads(response.body)
print("商品价格:", data.get("price"))
else:
print(f"❌ 失败：{response.status}")
print(response.body)
return 0
if __name__ == "__main__":
exit(main())
```
---
## API 参考
### 配置
#### `create_env_config() -> EnvConfig`
从环境变量创建配置（自动加载 `~/.openclaw/.env`）。
| 配置项 | 默认值 | 说明 |
|--------|--------|------|
| `AUTH_BASE` | `https://api-gw-test.yuanwei-lnc.com` | 认证基础 URL |
| `CLIENT_KEY` | **必需** | 客户端密钥（从全局配置加载） |
| `AUTH_CACHE_DIR` | `/tmp/skill-auth-cache` | 缓存目录 |
| `AUTH_MIN_TTL_SEC` | `60` | 最小令牌 TTL（秒） |
### 令牌管理
#### `get_access_token(dry_run, config) -> str`
获取访问令牌（带缓存）。
#### `refresh_access_token(dry_run, config) -> str`
刷新访问令牌（绕过缓存）。
### API 请求
#### `request_api(method, url, auth_token, body) -> ApiResponse`
发送 HTTP 请求。
#### `request_api_with_auto_refresh(method, url, dry_run, config, body) -> ApiResponse`
发送 API 请求并自动刷新令牌。
---
## 多 Agent Skill 配置
对于多 Agent skill，可以在全局配置中添加：
```bash
# ~/.openclaw/.env
# 主配置
CLIENT_KEY=sk_ae28fc4e.xxx
# Agent 特定配置
AGENT_1_CLIENT_KEY=sk_agent1.xxx
AGENT_2_CLIENT_KEY=sk_agent2.xxx
# 或者使用统一配置
# 所有 agent 共享同一个 CLIENT_KEY
```
在代码中：
```python
import os
from python_auth_runtime import create_env_config
# 主配置
config = create_env_config()
# 或者访问特定 agent 的配置
agent1_key = os.getenv("AGENT_1_CLIENT_KEY", config.client_key)
```
---
## 环境变量列表
| 变量 | 默认值 | 说明 |
|------|--------|------|
| `CLIENT_KEY` | **必需** | 客户端密钥 |
| `AUTH_BASE` | `https://api-gw-test.yuanwei-lnc.com` | 认证基础 URL |
| `AUTH_CACHE_DIR` | `/tmp/skill-auth-cache` | 缓存目录 |
| `AUTH_MIN_TTL_SEC` | `60` | 最小令牌 TTL（秒） |
| `GEMINI_API_KEY` | - | Gemini API Key（用于翻译） |
| `ECOM_BASE` | - | 1688 API 基础 URL |
---
## 测试
```bash
cd /Users/xiaolongxia/Documents/ai-build-app/skills/excel-toolkit/python_auth_runtime
uv run python scripts/example_usage.py
```
---
## 项目结构
```
python_auth_runtime/
├── src/python_auth_runtime/
│ └── __init__.py # 核心实现（自动加载全局配置）
├── scripts/
│ ├── load_env.py # .env 加载工具（可选使用）
│ └── example_usage.py # 使用示例
├── pyproject.toml
└── README.md
```
---
## 许可证
MIT

19
python_auth_runtime/pyproject.toml
View File

@ -1,19 +0,0 @@
[project]
name = "clawd-auth-runtime-py"
version = "1.0.0"
description = "Python 版本的 OpenClaw Auth Runtime"
readme = "README.md"
requires-python = ">=3.9"
dependencies = [
"requests>=2.28.0",
]
[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"
[tool.setuptools.packages.find]
where = ["src"]
[tool.uv]
index-url = "https://pypi.tuna.tsinghua.edu.cn/simple"

54
python_auth_runtime/scripts/example_usage.py
View File

@ -1,54 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
示例：如何在具体 skill 中使用 auth_runtime
参考：~/clawd/skills/1688-product-master/scripts/run.ts
"""
import sys
from pathlib import Path
# 添加当前目录到路径
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
# 步骤 1: 加载 .env.local
from load_env import load_env
load_env() # 加载 .env.local 或 .env
# 步骤 2: 导入 auth_runtime
from python_auth_runtime import create_env_config, request_api_with_auto_refresh
def main():
print("=" * 60)
print("Auth Runtime 使用示例")
print("=" * 60)
# 步骤 3: 创建配置（从环境变量读取）
config = create_env_config()
print(f"\n✓ 配置创建成功")
print(f" AUTH_BASE: {config.auth_base}")
print(f" CLIENT_KEY: {config.client_key[:10]}..." if config.client_key else " CLIENT_KEY: <empty>")
# 步骤 4: 调用 API
print(f"\n【测试】获取访问令牌...")
try:
response = request_api_with_auto_refresh(
method="POST",
url=f"{config.auth_base}/auth/skill-credit/session",
dry_run=False,
config=config,
body={"clientKey": config.client_key} if config.client_key else {},
)
print(f"✓ 响应状态：{response.status}")
if response.status == 200:
print(f" 响应体：{response.body[:100]}...")
else:
print(f" 响应体：{response.body}")
except Exception as e:
print(f"❌ 失败：{e}")
print("\n" + "=" * 60)
if __name__ == "__main__":
main()

90
python_auth_runtime/scripts/load_env.py
View File

@ -1,90 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
加载 .env.local 或 .env 文件到环境变量
在具体 skill 中使用：
from load_env import load_env
load_env() # 加载 .env.local 或 .env
参考：~/clawd/skills/1688-product-master/scripts/run.ts 中的 loadEnvLocal()
"""
import os
from pathlib import Path
def load_env(env_path: Path | None = None) -> bool:
"""
加载 .env.local 或 .env 文件到环境变量
优先级：
1. 指定的 env_path
2. 当前目录的 .env.local
3. 当前目录的 .env
4. 父目录的 .env.local
5. 父目录的 .env
Returns:
bool: 是否成功加载
"""
if env_path is None:
# 查找 .env 文件
possible_paths = [
Path.cwd() / ".env.local",
Path.cwd() / ".env",
Path.cwd().parent / ".env.local",
Path.cwd().parent / ".env",
]
for p in possible_paths:
if p.exists():
env_path = p
break
if env_path is None or not env_path.exists():
return False
print(f"📄 加载环境变量：{env_path}")
with open(env_path, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
# 跳过空行和注释
if not line or line.startswith("#"):
continue
# 解析 KEY=VALUE
eq_index = line.find("=")
if eq_index <= 0:
continue
key = line[:eq_index].strip()
value = line[eq_index + 1:].strip()
# 去除引号
if (value.startswith('"') and value.endswith('"')) or \
(value.startswith("'") and value.endswith("'")):
value = value[1:-1]
# 只设置尚未存在的环境变量（命令行参数优先级更高）
if key not in os.environ:
os.environ[key] = value
print(f" ✓ {key} = {value[:10]}..." if len(str(value)) > 10 else f" ✓ {key} = {value}")
return True
if __name__ == "__main__":
import sys
env_file = Path(sys.argv[1]) if len(sys.argv) > 1 else None
if load_env(env_file):
print("\n✅ 环境变量加载成功")
print(f"\n当前环境变量:")
for key in ["CLIENT_KEY", "AUTH_BASE", "AUTH_CACHE_DIR"]:
value = os.getenv(key, "<not set>")
print(f" {key} = {value[:10]}..." if len(str(value)) > 10 else f" {key} = {value}")
else:
print("❌ 未找到 .env 文件")
sys.exit(1)

372
python_auth_runtime/src/python_auth_runtime/__init__.py
View File

@ -1,372 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
@clawd/auth-runtime-py - Python 版本的 OpenClaw Auth Runtime
基于 ~/clawd/skills/_shared/auth-runtime 的 TypeScript 实现，
提供 Python 版本的客户端密钥认证。
配置加载优先级：
1. 环境变量（最高优先级）
2. 全局配置文件 ~/.openclaw/.env
3. 默认值
使用方式：
from python_auth_runtime import create_env_config, get_access_token, request_api_with_auto_refresh
config = create_env_config()
token = get_access_token(dry_run=False, config=config)
response = request_api_with_auto_refresh(
method="POST",
url=f"{config.auth_base}/your-endpoint",
dry_run=False,
config=config,
body={"key": "value"},
)
"""
import hashlib
import json
import os
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Optional
import requests # type: ignore
@dataclass
class EnvConfig:
"""环境配置"""
auth_base: str = "https://api-gw-test.yuanwei-lnc.com"
client_key: str = ""
auth_cache_dir: str = "/tmp/skill-auth-cache"
auth_min_ttl_sec: int = 60
@dataclass
class SessionResponse:
"""会话响应"""
access_token: str
hook_url: Optional[str] = None
hook_token: Optional[str] = None
expires_in: int = 900
@dataclass
class CachedTokenData:
"""缓存的令牌数据"""
access_token: str
expires_at: float
@dataclass
class ApiResponse:
"""HTTP 响应"""
status: int
body: str
headers: dict[str, str]
# 可重试的状态码
RETRYABLE_STATUS = {401, 403}
# 可重试的响应体标记
RETRYABLE_BODY_MARKERS = [
'session not found or expired',
'invalid or expired token',
'unauthorized',
'client key expired',
'client key revoked',
]
# 全局配置文件路径
GLOBAL_ENV_PATHS = [
Path.home() / ".openclaw" / ".env",
Path.home() / ".clawd" / ".env",
Path.home() / "clawd" / ".env",
]
def load_global_env() -> bool:
"""
从全局配置文件加载环境变量
加载优先级：
1. ~/.openclaw/.env
2. ~/.clawd/.env
3. ~/clawd/.env
只加载尚未存在的环境变量（命令行参数优先级更高）
Returns:
bool: 是否成功加载
"""
for env_path in GLOBAL_ENV_PATHS:
if env_path.exists():
print(f"📄 加载全局配置：{env_path}")
with open(env_path, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line or line.startswith("#"):
continue
eq_index = line.find("=")
if eq_index > 0:
key = line[:eq_index].strip()
value = line[eq_index + 1:].strip()
# 去除引号
if (value.startswith('"') and value.endswith('"')) or \
(value.startswith("'") and value.endswith("'")):
value = value[1:-1]
# 只加载尚未存在的环境变量
if key not in os.environ:
os.environ[key] = value
print(f" ✓ {key} = {value[:10]}..." if len(str(value)) > 10 else f" ✓ {key} = {value}")
return True
return False
def create_env_config() -> EnvConfig:
"""
从环境变量创建配置
加载优先级：
1. 环境变量（最高优先级）
2. 全局配置文件 ~/.openclaw/.env
3. 默认值
环境变量：
- AUTH_BASE: 认证基础 URL（默认：https://api-gw-test.yuanwei-lnc.com）
- CLIENT_KEY: 客户端密钥（必需）
- AUTH_CACHE_DIR: 缓存目录（默认：/tmp/skill-auth-cache）
- AUTH_MIN_TTL_SEC: 最小令牌 TTL 秒数（默认：60）
"""
# 先尝试加载全局配置
load_global_env()
auth_base = os.getenv("AUTH_BASE", "https://api-gw-test.yuanwei-lnc.com").rstrip("/")
client_key = os.getenv("CLIENT_KEY", "")
auth_cache_dir = os.getenv("AUTH_CACHE_DIR", "/tmp/skill-auth-cache")
auth_min_ttl_sec = int(os.getenv("AUTH_MIN_TTL_SEC", "60"))
return EnvConfig(
auth_base=auth_base,
client_key=client_key,
auth_cache_dir=auth_cache_dir,
auth_min_ttl_sec=auth_min_ttl_sec,
)
def get_cache_file(auth_base: str, client_key: str, cache_dir: str) -> Path:
"""获取缓存文件路径"""
cache_path = Path(cache_dir)
cache_path.mkdir(parents=True, exist_ok=True)
key_str = f"{auth_base}:{client_key}"
hash_value = hashlib.sha256(key_str.encode()).hexdigest()[:16]
return cache_path / f"token_{hash_value}.json"
def read_cached_token(cache_file: Path, min_ttl_sec: int) -> Optional[str]:
"""读取缓存的令牌"""
if not cache_file.exists():
return None
try:
with open(cache_file, "r", encoding="utf-8") as f:
data = json.load(f)
cached = CachedTokenData(
access_token=data["access_token"],
expires_at=data["expires_at"],
)
if time.time() + min_ttl_sec < cached.expires_at:
return cached.access_token
else:
cache_file.unlink(missing_ok=True)
return None
except (json.JSONDecodeError, KeyError, Exception):
cache_file.unlink(missing_ok=True)
return None
def write_cache(cache_file: Path, session: SessionResponse) -> None:
"""写入缓存"""
cache_file.parent.mkdir(parents=True, exist_ok=True)
data = {
"access_token": session.access_token,
"expires_at": time.time() + session.expires_in,
"expires_in": session.expires_in,
}
with open(cache_file, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2)
def delete_cache(cache_file: Path) -> None:
"""删除缓存"""
cache_file.unlink(missing_ok=True)
def fetch_session_json(dry_run: bool, config: EnvConfig) -> SessionResponse:
"""从认证端点获取会话 JSON"""
if dry_run:
return SessionResponse(
access_token="<dry-run-token>",
hook_url="<dry-run-hook-url>",
hook_token="<dry-run-hook-token>",
expires_in=900,
)
if not config.client_key:
raise ValueError("CLIENT_KEY is required. Please set it in ~/.openclaw/.env or environment variable.")
payload = {"clientKey": config.client_key}
url = f"{config.auth_base}/auth/skill-credit/session"
response = requests.post(url, json=payload, timeout=30)
if response.status_code < 200 or response.status_code >= 300:
raise RuntimeError(
f"Auth session request failed: HTTP {response.status_code} - {response.text}"
)
session_data = response.json()
if not session_data.get("accessToken"):
raise RuntimeError(f"Missing accessToken in session response: {response.text}")
return SessionResponse(
access_token=session_data["accessToken"],
hook_url=session_data.get("hookUrl"),
hook_token=session_data.get("hookToken"),
expires_in=session_data.get("expiresIn", 900),
)
def get_access_token(dry_run: bool, config: EnvConfig) -> str:
"""获取访问令牌（带缓存）"""
if dry_run:
return "<dry-run-token>"
if not config.client_key:
raise ValueError("CLIENT_KEY is required. Please set it in ~/.openclaw/.env or environment variable.")
cache_file = get_cache_file(config.auth_base, config.client_key, config.auth_cache_dir)
cached_token = read_cached_token(cache_file, config.auth_min_ttl_sec)
if cached_token:
return cached_token
session = fetch_session_json(dry_run, config)
write_cache(cache_file, session)
return session.access_token
def refresh_access_token(dry_run: bool, config: EnvConfig) -> str:
"""刷新访问令牌（绕过缓存）"""
if dry_run:
return "<dry-run-token>"
if not config.client_key:
raise ValueError("CLIENT_KEY is required")
cache_file = get_cache_file(config.auth_base, config.client_key, config.auth_cache_dir)
delete_cache(cache_file)
return get_access_token(dry_run, config)
def is_retryable_session_error(response: ApiResponse) -> bool:
"""检查响应是否表示会话过期/无效"""
if response.status not in RETRYABLE_STATUS:
return False
body = (response.body or "").lower()
if not body:
return True
return any(marker in body for marker in RETRYABLE_BODY_MARKERS)
def request_api(
method: str,
url: str,
auth_token: Optional[str] = None,
body: Optional[dict[str, Any]] = None,
timeout: int = 30,
) -> ApiResponse:
"""发送 HTTP 请求"""
headers = {"Content-Type": "application/json"}
if auth_token:
headers["Authorization"] = f"Bearer {auth_token}"
try:
if method.upper() == "GET":
response = requests.get(url, headers=headers, timeout=timeout)
elif method.upper() == "POST":
response = requests.post(url, headers=headers, json=body, timeout=timeout)
elif method.upper() == "PUT":
response = requests.put(url, headers=headers, json=body, timeout=timeout)
elif method.upper() == "DELETE":
response = requests.delete(url, headers=headers, timeout=timeout)
else:
raise ValueError(f"Unsupported HTTP method: {method}")
return ApiResponse(
status=response.status_code,
body=response.text,
headers=dict(response.headers),
)
except requests.exceptions.RequestException as e:
return ApiResponse(
status=0,
body=str(e),
headers={},
)
def request_api_with_auto_refresh(
method: str,
url: str,
dry_run: bool,
config: EnvConfig,
body: Optional[dict[str, Any]] = None,
access_token: Optional[str] = None,
) -> ApiResponse:
"""发送 API 请求并自动刷新令牌"""
token = access_token or get_access_token(dry_run, config)
first = request_api(method, url, token, body)
if not is_retryable_session_error(first):
return first
fresh_token = refresh_access_token(dry_run, config)
return request_api(method, url, fresh_token, body)
__version__ = "1.0.0"
__all__ = [
"EnvConfig",
"SessionResponse",
"ApiResponse",
"load_global_env",
"create_env_config",
"get_access_token",
"refresh_access_token",
"fetch_session_json",
"request_api",
"request_api_with_auto_refresh",
"is_retryable_session_error",
]

1652
uv.lock
View File

File diff suppressed because it is too large Load Diff