refactor: 将 auth runtime 重构为可复用 Python 包
模式参考 ~/clawd/skills/_shared/auth-runtime (TypeScript):
- 创建 python_auth_runtime/ 作为独立 Python 包
- 其他 skill 可以通过 uv pip install 引用
- 支持三种使用方式：
1. 本地包安装：uv pip install /path/to/python_auth_runtime
2. 文件依赖：pyproject.toml 中引用
3. 复制源码：直接复制 src/python_auth_runtime
包结构：
python_auth_runtime/
├── pyproject.toml # 包配置
├── README.md # 使用文档
└── src/python_auth_runtime/
└── __init__.py # 核心实现
功能：
- CLIENT_KEY 认证
- 令牌缓存（可配置 TTL）
- 自动刷新过期令牌
- 401/403 自动重试
- 从环境变量加载配置
This commit is contained in:
ivanberry
parent
2eff4a6033
commit
98ca8a3965
|
|
@ -7,6 +7,9 @@ dependencies = [
|
|||
"pandas>=1.5.0",
|
||||
"openpyxl>=3.0.0",
|
||||
"google-generativeai>=0.8.0",
|
||||
"requests>=2.28.0",
|
||||
# 引用本地 auth-runtime 包
|
||||
"python-auth-runtime-py @ file:///${PROJECT_ROOT}/python_auth_runtime",
|
||||
]
|
||||
|
||||
[tool.uv]
|
||||
|
|
|
|||
|
|
@ -0,0 +1,163 @@
|
|||
# @clawd/auth-runtime-py
|
||||
|
||||
Python 版本的 OpenClaw Auth Runtime，基于 `~/clawd/skills/_shared/auth-runtime` 的 TypeScript 实现。
|
||||
|
||||
## 安装
|
||||
|
||||
### 方式 1: 作为本地包安装（推荐）
|
||||
|
||||
```bash
|
||||
# 在具体 skill 的目录中
|
||||
uv pip install /path/to/python_auth_runtime
|
||||
```
|
||||
|
||||
### 方式 2: 作为文件依赖
|
||||
|
||||
在 `pyproject.toml` 中添加：
|
||||
|
||||
```toml
|
||||
[project]
|
||||
dependencies = [
|
||||
"python-auth-runtime-py @ file:///path/to/python_auth_runtime",
|
||||
]
|
||||
```
|
||||
|
||||
### 方式 3: 复制源码
|
||||
|
||||
直接复制 `src/python_auth_runtime` 到你的项目中：
|
||||
|
||||
```bash
|
||||
cp -r /path/to/python_auth_runtime/src/python_auth_runtime ./your-skill/
|
||||
```
|
||||
|
||||
## 使用方式
|
||||
|
||||
### 1. 设置环境变量
|
||||
|
||||
```bash
|
||||
export CLIENT_KEY="your-client-key"
|
||||
export AUTH_BASE="https://api-gw-test.yuanwei-lnc.com" # 可选
|
||||
```
|
||||
|
||||
### 2. 导入并使用
|
||||
|
||||
```python
|
||||
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)
|
||||
|
||||
# 发送 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:
|
||||
import json
|
||||
data = json.loads(response.body)
|
||||
print("商品价格:", data["price"])
|
||||
```
|
||||
|
||||
## 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`
|
||||
|
||||
获取访问令牌（带缓存）
|
||||
|
||||
#### `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 请求并自动刷新令牌
|
||||
|
||||
## 与 TypeScript 版本对比
|
||||
|
||||
| 特性 | TypeScript | Python |
|
||||
|------|-----------|--------|
|
||||
| 包名 | `@clawd/auth-runtime` | `@clawd/auth-runtime-py` |
|
||||
| 安装方式 | `bun add file:../_shared/auth-runtime` | `uv pip install /path/to/python_auth_runtime` |
|
||||
| 导入 | `import { ... } from '@clawd/auth-runtime'` | `from python_auth_runtime import ...` |
|
||||
| HTTP 客户端 | `fetch()` | `requests` |
|
||||
|
||||
## 示例项目
|
||||
|
||||
```python
|
||||
# your-skill/main.py
|
||||
from python_auth_runtime import create_env_config, request_api_with_auto_refresh
|
||||
import os
|
||||
|
||||
def main():
|
||||
# 从环境变量加载 CLIENT_KEY
|
||||
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:
|
||||
print("成功:", response.body)
|
||||
else:
|
||||
print("失败:", response.status, response.body)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
```
|
||||
|
||||
## 环境变量配置
|
||||
|
||||
在项目根目录创建 `.env` 文件（不提交到 git）：
|
||||
|
||||
```bash
|
||||
# .env
|
||||
CLIENT_KEY=your-client-key-here
|
||||
AUTH_BASE=https://api-gw-test.yuanwei-lnc.com
|
||||
```
|
||||
|
||||
然后在代码中加载：
|
||||
|
||||
```python
|
||||
from dotenv import load_dotenv
|
||||
load_dotenv() # 加载 .env 文件到环境变量
|
||||
|
||||
from python_auth_runtime import create_env_config
|
||||
config = create_env_config()
|
||||
```
|
||||
|
||||
## 许可证
|
||||
|
||||
MIT
|
||||
|
|
@ -0,0 +1,19 @@
|
|||
[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"
|
||||
|
|
@ -1,14 +1,24 @@
|
|||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Python 版本的 OpenClaw Auth Runtime
|
||||
基于 ~/clawd/skills/_shared/auth-runtime 的 TypeScript 实现
|
||||
@clawd/auth-runtime-py - Python 版本的 OpenClaw Auth Runtime
|
||||
|
||||
功能：
|
||||
- 使用 CLIENT_KEY 获取访问令牌
|
||||
- 支持令牌缓存（可配置 TTL）
|
||||
- 自动刷新过期令牌
|
||||
- 401/403 自动重试
|
||||
基于 ~/clawd/skills/_shared/auth-runtime 的 TypeScript 实现，
|
||||
提供 Python 版本的客户端密钥认证。
|
||||
|
||||
使用方式：
|
||||
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
|
||||
|
|
@ -92,15 +102,10 @@ def create_env_config() -> EnvConfig:
|
|||
|
||||
|
||||
def get_cache_file(auth_base: str, client_key: str, cache_dir: str) -> Path:
|
||||
"""
|
||||
获取缓存文件路径
|
||||
|
||||
使用 auth_base 和 client_key 的哈希值生成唯一的缓存文件名
|
||||
"""
|
||||
"""获取缓存文件路径"""
|
||||
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]
|
||||
|
||||
|
|
@ -108,11 +113,7 @@ def get_cache_file(auth_base: str, client_key: str, cache_dir: str) -> Path:
|
|||
|
||||
|
||||
def read_cached_token(cache_file: Path, min_ttl_sec: int) -> Optional[str]:
|
||||
"""
|
||||
读取缓存的令牌
|
||||
|
||||
如果令牌存在且剩余 TTL 大于最小 TTL，则返回令牌
|
||||
"""
|
||||
"""读取缓存的令牌"""
|
||||
if not cache_file.exists():
|
||||
return None
|
||||
|
||||
|
|
@ -125,26 +126,19 @@ def read_cached_token(cache_file: Path, min_ttl_sec: int) -> Optional[str]:
|
|||
expires_at=data["expires_at"],
|
||||
)
|
||||
|
||||
# 检查是否过期（考虑最小 TTL）
|
||||
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 = {
|
||||
|
|
@ -163,11 +157,7 @@ def delete_cache(cache_file: Path) -> None:
|
|||
|
||||
|
||||
def fetch_session_json(dry_run: bool, config: EnvConfig) -> SessionResponse:
|
||||
"""
|
||||
从认证端点获取会话 JSON
|
||||
|
||||
使用 CLIENT_KEY 请求访问令牌
|
||||
"""
|
||||
"""从认证端点获取会话 JSON"""
|
||||
if dry_run:
|
||||
return SessionResponse(
|
||||
access_token="<dry-run-token>",
|
||||
|
|
@ -179,7 +169,6 @@ def fetch_session_json(dry_run: bool, config: EnvConfig) -> SessionResponse:
|
|||
if not config.client_key:
|
||||
raise ValueError("CLIENT_KEY is required")
|
||||
|
||||
# 请求认证端点
|
||||
payload = {"clientKey": config.client_key}
|
||||
url = f"{config.auth_base}/auth/skill-credit/session"
|
||||
|
||||
|
|
@ -204,13 +193,7 @@ def fetch_session_json(dry_run: bool, config: EnvConfig) -> SessionResponse:
|
|||
|
||||
|
||||
def get_access_token(dry_run: bool, config: EnvConfig) -> str:
|
||||
"""
|
||||
获取访问令牌（带缓存）
|
||||
|
||||
1. 检查缓存
|
||||
2. 如果缓存有效，返回缓存的令牌
|
||||
3. 否则请求新令牌并缓存
|
||||
"""
|
||||
"""获取访问令牌（带缓存）"""
|
||||
if dry_run:
|
||||
return "<dry-run-token>"
|
||||
|
||||
|
|
@ -221,23 +204,16 @@ def get_access_token(dry_run: bool, config: EnvConfig) -> str:
|
|||
cached_token = read_cached_token(cache_file, config.auth_min_ttl_sec)
|
||||
|
||||
if cached_token:
|
||||
print(f"✓ 使用缓存的访问令牌")
|
||||
return cached_token
|
||||
|
||||
print(f"🔑 请求新的访问令牌...")
|
||||
session = fetch_session_json(dry_run, config)
|
||||
write_cache(cache_file, session)
|
||||
print(f"✓ 令牌已缓存到：{cache_file}")
|
||||
|
||||
return session.access_token
|
||||
|
||||
|
||||
def refresh_access_token(dry_run: bool, config: EnvConfig) -> str:
|
||||
"""
|
||||
刷新访问令牌（绕过缓存）
|
||||
|
||||
删除缓存并重新请求新令牌
|
||||
"""
|
||||
"""刷新访问令牌（绕过缓存）"""
|
||||
if dry_run:
|
||||
return "<dry-run-token>"
|
||||
|
||||
|
|
@ -246,17 +222,12 @@ def refresh_access_token(dry_run: bool, config: EnvConfig) -> str:
|
|||
|
||||
cache_file = get_cache_file(config.auth_base, config.client_key, config.auth_cache_dir)
|
||||
delete_cache(cache_file)
|
||||
print(f"🔄 刷新访问令牌...")
|
||||
|
||||
return get_access_token(dry_run, config)
|
||||
|
||||
|
||||
def is_retryable_session_error(response: ApiResponse) -> bool:
|
||||
"""
|
||||
检查响应是否表示会话过期/无效
|
||||
|
||||
如果是 401/403 或响应体包含特定标记，则可以重试
|
||||
"""
|
||||
"""检查响应是否表示会话过期/无效"""
|
||||
if response.status not in RETRYABLE_STATUS:
|
||||
return False
|
||||
|
||||
|
|
@ -274,19 +245,7 @@ def request_api(
|
|||
body: Optional[dict[str, Any]] = None,
|
||||
timeout: int = 30,
|
||||
) -> ApiResponse:
|
||||
"""
|
||||
发送 HTTP 请求
|
||||
|
||||
Args:
|
||||
method: HTTP 方法（GET, POST, PUT, DELETE 等）
|
||||
url: 请求 URL
|
||||
auth_token: 访问令牌（可选）
|
||||
body: 请求体（可选，自动序列化为 JSON）
|
||||
timeout: 超时时间（秒）
|
||||
|
||||
Returns:
|
||||
ApiResponse: 响应对象
|
||||
"""
|
||||
"""发送 HTTP 请求"""
|
||||
headers = {"Content-Type": "application/json"}
|
||||
|
||||
if auth_token:
|
||||
|
|
@ -296,13 +255,9 @@ def request_api(
|
|||
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
|
||||
)
|
||||
response = requests.post(url, headers=headers, json=body, timeout=timeout)
|
||||
elif method.upper() == "PUT":
|
||||
response = requests.put(
|
||||
url, headers=headers, json=body, timeout=timeout
|
||||
)
|
||||
response = requests.put(url, headers=headers, json=body, timeout=timeout)
|
||||
elif method.upper() == "DELETE":
|
||||
response = requests.delete(url, headers=headers, timeout=timeout)
|
||||
else:
|
||||
|
|
@ -330,66 +285,28 @@ def request_api_with_auto_refresh(
|
|||
body: Optional[dict[str, Any]] = None,
|
||||
access_token: Optional[str] = None,
|
||||
) -> ApiResponse:
|
||||
"""
|
||||
发送 API 请求并自动刷新令牌
|
||||
|
||||
1. 使用当前令牌发送请求
|
||||
2. 如果是 401/403 错误，刷新令牌并重试一次
|
||||
"""
|
||||
"""发送 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
|
||||
|
||||
# 刷新令牌并重试
|
||||
print(f"⚠️ 检测到会话过期，刷新令牌后重试...")
|
||||
fresh_token = refresh_access_token(dry_run, config)
|
||||
return request_api(method, url, fresh_token, body)
|
||||
|
||||
|
||||
# 便捷函数：从 .env 文件加载 CLIENT_KEY
|
||||
def load_client_key_from_env(env_path: Optional[Path] = None) -> str:
|
||||
"""
|
||||
从 .env 文件加载 CLIENT_KEY
|
||||
|
||||
优先级：
|
||||
1. 环境变量 CLIENT_KEY
|
||||
2. .env 文件中的 CLIENT_KEY
|
||||
3. 抛出异常
|
||||
"""
|
||||
# 先检查环境变量
|
||||
client_key = os.getenv("CLIENT_KEY")
|
||||
if client_key:
|
||||
return client_key
|
||||
|
||||
# 查找 .env 文件
|
||||
if env_path is None:
|
||||
possible_paths = [
|
||||
Path(".env"),
|
||||
Path.cwd() / ".env",
|
||||
Path(__file__).parent.parent / ".env",
|
||||
]
|
||||
for p in possible_paths:
|
||||
if p.exists():
|
||||
env_path = p
|
||||
break
|
||||
|
||||
if env_path and env_path.exists():
|
||||
with open(env_path, "r", encoding="utf-8") as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if not line or line.startswith("#"):
|
||||
continue
|
||||
if "=" in line:
|
||||
key, value = line.split("=", 1)
|
||||
if key.strip() == "CLIENT_KEY":
|
||||
return value.strip()
|
||||
|
||||
raise ValueError(
|
||||
"CLIENT_KEY not found. Please set CLIENT_KEY environment variable "
|
||||
"or add it to .env file"
|
||||
)
|
||||
__version__ = "1.0.0"
|
||||
__all__ = [
|
||||
"EnvConfig",
|
||||
"SessionResponse",
|
||||
"ApiResponse",
|
||||
"create_env_config",
|
||||
"get_access_token",
|
||||
"refresh_access_token",
|
||||
"fetch_session_json",
|
||||
"request_api",
|
||||
"request_api_with_auto_refresh",
|
||||
"is_retryable_session_error",
|
||||
]
|
||||
|
|
@ -1,75 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
测试 OpenClaw Auth Runtime
|
||||
"""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# 添加当前目录到路径
|
||||
sys.path.insert(0, str(Path(__file__).parent))
|
||||
|
||||
from auth_runtime import (
|
||||
create_env_config,
|
||||
get_access_token,
|
||||
request_api_with_auto_refresh,
|
||||
load_client_key_from_env,
|
||||
)
|
||||
|
||||
|
||||
def main():
|
||||
print("=" * 60)
|
||||
print("OpenClaw Auth Runtime 测试")
|
||||
print("=" * 60)
|
||||
|
||||
# 方式 1: 从 .env 加载 CLIENT_KEY
|
||||
try:
|
||||
client_key = load_client_key_from_env()
|
||||
print(f"✓ 从 .env 文件加载 CLIENT_KEY: {client_key[:10]}...")
|
||||
except ValueError as e:
|
||||
print(f"❌ {e}")
|
||||
print("\n请在 .env 文件中设置 CLIENT_KEY")
|
||||
print("或设置环境变量：export CLIENT_KEY=your-key")
|
||||
return 1
|
||||
|
||||
# 创建配置
|
||||
config = create_env_config()
|
||||
print(f"✓ 配置创建成功")
|
||||
print(f" AUTH_BASE: {config.auth_base}")
|
||||
print(f" AUTH_CACHE_DIR: {config.auth_cache_dir}")
|
||||
|
||||
# 获取访问令牌
|
||||
print("\n【测试 1】获取访问令牌...")
|
||||
try:
|
||||
token = get_access_token(dry_run=False, config=config)
|
||||
print(f"✓ 获取成功：{token[:20]}...")
|
||||
except Exception as e:
|
||||
print(f"❌ 失败：{e}")
|
||||
return 1
|
||||
|
||||
# 测试 API 请求
|
||||
print("\n【测试 2】测试 API 请求...")
|
||||
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},
|
||||
)
|
||||
print(f"✓ 响应状态：{response.status}")
|
||||
print(f" 响应体：{response.body[:100]}...")
|
||||
except Exception as e:
|
||||
print(f"❌ 失败：{e}")
|
||||
|
||||
print("\n" + "=" * 60)
|
||||
print("✅ 测试完成")
|
||||
print("=" * 60)
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Loading…
Reference in New Issue