mirror of
https://github.com/cft0808/edict.git
synced 2026-09-06 08:58:42 +08:00
709e163932
新增远程 skill 管理系统，支持从 GitHub/HTTPS URL 获取: - API 端点: add-remote-skill, remote-skills-list, update-remote-skill, remove-remote-skill - CLI 工具: scripts/skill_manager.py (add-remote, list-remote, update-remote, remove-remote, import-official-hub) - 文档: remote-skills-guide.md 和 remote-skills-quickstart.md - 功能: 安全验证、YAML 校验、版本管理、多渠道操作
317 lines
11 KiB
Python
317 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
三省六部 · Skill 管理工具
|
||
支持从本地或远程 URL 添加、更新、查看和移除 skills
|
||
|
||
Usage:
|
||
python3 scripts/skill_manager.py add-remote --agent zhongshu --name code_review \\
|
||
--source https://raw.githubusercontent.com/org/skills/main/code_review/SKILL.md \\
|
||
--description "代码审查"
|
||
|
||
python3 scripts/skill_manager.py list-remote
|
||
|
||
python3 scripts/skill_manager.py update-remote --agent zhongshu --name code_review
|
||
|
||
python3 scripts/skill_manager.py remove-remote --agent zhongshu --name code_review
|
||
|
||
python3 scripts/skill_manager.py import-official-hub --agents zhongshu,menxia,shangshu
|
||
"""
|
||
import sys
|
||
import json
|
||
import pathlib
|
||
import argparse
|
||
import urllib.request
|
||
import urllib.error
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).parent))
|
||
from utils import now_iso, safe_name, read_json
|
||
|
||
OCLAW_HOME = Path.home() / '.openclaw'
|
||
|
||
|
||
def _download_file(url: str, timeout: int = 10) -> str:
|
||
"""从 URL 下载文件内容（文本格式）"""
|
||
try:
|
||
req = urllib.request.Request(url, headers={'User-Agent': 'OpenClaw-SkillManager/1.0'})
|
||
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
||
content = resp.read(10 * 1024 * 1024) # 最多 10MB
|
||
return content.decode('utf-8')
|
||
except urllib.error.HTTPError as e:
|
||
raise Exception(f'HTTP {e.code}: {e.reason}')
|
||
except Exception as e:
|
||
raise Exception(f'{type(e).__name__}: {e}')
|
||
|
||
|
||
def _compute_checksum(content: str) -> str:
|
||
"""计算内容的简单校验和"""
|
||
import hashlib
|
||
return hashlib.sha256(content.encode()).hexdigest()[:16]
|
||
|
||
|
||
def add_remote(agent_id: str, name: str, source_url: str, description: str = '') -> bool:
|
||
"""从远程 URL 为 Agent 添加 skill"""
|
||
if not safe_name(agent_id) or not safe_name(name):
|
||
print(f'❌ 错误：agent_id 或 skill 名称含非法字符')
|
||
return False
|
||
|
||
# 设置 workspace
|
||
workspace = OCLAW_HOME / f'workspace-{agent_id}' / 'skills' / name
|
||
workspace.mkdir(parents=True, exist_ok=True)
|
||
skill_md = workspace / 'SKILL.md'
|
||
|
||
# 下载文件
|
||
print(f'⏳ 正在从 {source_url} 下载...')
|
||
try:
|
||
content = _download_file(source_url)
|
||
except Exception as e:
|
||
print(f'❌ 下载失败：{e}')
|
||
return False
|
||
|
||
# 基础验证
|
||
if not content.startswith('---'):
|
||
print(f'❌ 文件格式无效：缺少 YAML frontmatter')
|
||
return False
|
||
|
||
# 保存 SKILL.md
|
||
skill_md.write_text(content)
|
||
|
||
# 保存源信息
|
||
source_info = {
|
||
'skillName': name,
|
||
'sourceUrl': source_url,
|
||
'description': description,
|
||
'addedAt': now_iso(),
|
||
'lastUpdated': now_iso(),
|
||
'checksum': _compute_checksum(content),
|
||
'status': 'valid',
|
||
}
|
||
source_json = workspace / '.source.json'
|
||
source_json.write_text(json.dumps(source_info, ensure_ascii=False, indent=2))
|
||
|
||
print(f'✅ 技能 {name} 已添加到 {agent_id}')
|
||
print(f' 路径: {skill_md}')
|
||
print(f' 大小: {len(content)} 字节')
|
||
return True
|
||
|
||
|
||
def list_remote() -> bool:
|
||
"""列出所有已添加的远程 skills"""
|
||
if not OCLAW_HOME.exists():
|
||
print('❌ OCLAW_HOME 不存在')
|
||
return False
|
||
|
||
remote_skills = []
|
||
|
||
for ws_dir in OCLAW_HOME.glob('workspace-*'):
|
||
agent_id = ws_dir.name.replace('workspace-', '')
|
||
skills_dir = ws_dir / 'skills'
|
||
if not skills_dir.exists():
|
||
continue
|
||
|
||
for skill_dir in skills_dir.iterdir():
|
||
if not skill_dir.is_dir():
|
||
continue
|
||
skill_name = skill_dir.name
|
||
source_json = skill_dir / '.source.json'
|
||
|
||
if not source_json.exists():
|
||
continue
|
||
|
||
try:
|
||
source_info = json.loads(source_json.read_text())
|
||
remote_skills.append({
|
||
'agent': agent_id,
|
||
'skill': skill_name,
|
||
'source': source_info.get('sourceUrl', 'N/A'),
|
||
'desc': source_info.get('description', ''),
|
||
'added': source_info.get('addedAt', 'N/A'),
|
||
})
|
||
except Exception:
|
||
pass
|
||
|
||
if not remote_skills:
|
||
print('📭 暂无远程 skills')
|
||
return True
|
||
|
||
print(f'📋 共 {len(remote_skills)} 个远程 skills：\n')
|
||
print(f'{"Agent":<12} | {"Skill 名称":<20} | {"描述":<30} | 添加时间')
|
||
print('-' * 100)
|
||
|
||
for sk in remote_skills:
|
||
desc = (sk['desc'] or sk['source'])[:30].ljust(30)
|
||
print(f"{sk['agent']:<12} | {sk['skill']:<20} | {desc} | {sk['added'][:10]}")
|
||
|
||
print()
|
||
return True
|
||
|
||
|
||
def update_remote(agent_id: str, name: str) -> bool:
|
||
"""更新远程 skill 为最新版本"""
|
||
if not safe_name(agent_id) or not safe_name(name):
|
||
print(f'❌ 错误：agent_id 或 skill 名称含非法字符')
|
||
return False
|
||
|
||
workspace = OCLAW_HOME / f'workspace-{agent_id}' / 'skills' / name
|
||
source_json = workspace / '.source.json'
|
||
|
||
if not source_json.exists():
|
||
print(f'❌ 技能不存在或不是远程 skill: {name}')
|
||
return False
|
||
|
||
try:
|
||
source_info = json.loads(source_json.read_text())
|
||
source_url = source_info.get('sourceUrl')
|
||
if not source_url:
|
||
print(f'❌ 无效的源 URL')
|
||
return False
|
||
|
||
# 重新下载
|
||
return add_remote(agent_id, name, source_url, source_info.get('description', ''))
|
||
except Exception as e:
|
||
print(f'❌ 更新失败：{e}')
|
||
return False
|
||
|
||
|
||
def remove_remote(agent_id: str, name: str) -> bool:
|
||
"""移除远程 skill"""
|
||
if not safe_name(agent_id) or not safe_name(name):
|
||
print(f'❌ 错误：agent_id 或 skill 名称含非法字符')
|
||
return False
|
||
|
||
workspace = OCLAW_HOME / f'workspace-{agent_id}' / 'skills' / name
|
||
source_json = workspace / '.source.json'
|
||
|
||
if not source_json.exists():
|
||
print(f'❌ 技能不存在或不是远程 skill: {name}')
|
||
return False
|
||
|
||
try:
|
||
import shutil
|
||
shutil.rmtree(workspace)
|
||
print(f'✅ 技能 {name} 已从 {agent_id} 移除')
|
||
return True
|
||
except Exception as e:
|
||
print(f'❌ 移除失败：{e}')
|
||
return False
|
||
|
||
|
||
OFFICIAL_SKILLS_HUB = {
|
||
'code_review': 'https://raw.githubusercontent.com/openclaw-ai/skills-hub/main/code_review/SKILL.md',
|
||
'api_design': 'https://raw.githubusercontent.com/openclaw-ai/skills-hub/main/api_design/SKILL.md',
|
||
'security_audit': 'https://raw.githubusercontent.com/openclaw-ai/skills-hub/main/security_audit/SKILL.md',
|
||
'data_analysis': 'https://raw.githubusercontent.com/openclaw-ai/skills-hub/main/data_analysis/SKILL.md',
|
||
'doc_generation': 'https://raw.githubusercontent.com/openclaw-ai/skills-hub/main/doc_generation/SKILL.md',
|
||
'test_framework': 'https://raw.githubusercontent.com/openclaw-ai/skills-hub/main/test_framework/SKILL.md',
|
||
}
|
||
|
||
SKILL_AGENT_MAPPING = {
|
||
'code_review': ('bingbu', 'xingbu', 'menxia'),
|
||
'api_design': ('bingbu', 'gongbu', 'menxia'),
|
||
'security_audit': ('xingbu', 'menxia'),
|
||
'data_analysis': ('hubu', 'menxia'),
|
||
'doc_generation': ('libu', 'menxia'),
|
||
'test_framework': ('gongbu', 'xingbu', 'menxia'),
|
||
}
|
||
|
||
|
||
def import_official_hub(agent_ids: list) -> bool:
|
||
"""从官方 Skills Hub 导入指定的 skills 到指定 agents。
|
||
如果未指定 agents，使用该 skill 的推荐 agents。
|
||
"""
|
||
if not agent_ids:
|
||
print('❌ 未指定 agent，使用推荐配置...\n')
|
||
for skill_name, recommended_agents in SKILL_AGENT_MAPPING.items():
|
||
agent_ids.extend(recommended_agents)
|
||
agent_ids = list(set(agent_ids))
|
||
|
||
total = 0
|
||
success = 0
|
||
|
||
for skill_name, url in OFFICIAL_SKILLS_HUB.items():
|
||
# 确定目标 agents
|
||
target_agents = agent_ids
|
||
if not agent_ids:
|
||
target_agents = SKILL_AGENT_MAPPING.get(skill_name, ['menxia'])
|
||
|
||
print(f'\n📥 正在导入 skill: {skill_name}')
|
||
print(f' 目标 agents: {", ".join(target_agents)}')
|
||
|
||
for agent_id in target_agents:
|
||
total += 1
|
||
if add_remote(agent_id, skill_name, url, f'官方 skill：{skill_name}'):
|
||
success += 1
|
||
|
||
print(f'\n📊 导入完成：{success}/{total} 个 skills 成功')
|
||
return success == total
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description='三省六部 Skill 管理工具',
|
||
formatter_class=argparse.RawDescriptionHelpFormatter)
|
||
subparsers = parser.add_subparsers(dest='cmd', help='命令')
|
||
|
||
# add-remote
|
||
add_parser = subparsers.add_parser('add-remote', help='从远程 URL 添加 skill')
|
||
add_parser.add_argument('--agent', required=True, help='目标 Agent ID')
|
||
add_parser.add_argument('--name', required=True, help='Skill 内部名称')
|
||
add_parser.add_argument('--source', required=True, help='远程 URL 或本地路径')
|
||
add_parser.add_argument('--description', default='', help='Skill 描述')
|
||
|
||
# list-remote
|
||
subparsers.add_parser('list-remote', help='列出所有远程 skills')
|
||
|
||
# update-remote
|
||
update_parser = subparsers.add_parser('update-remote', help='更新远程 skill')
|
||
update_parser.add_argument('--agent', required=True, help='Agent ID')
|
||
update_parser.add_argument('--name', required=True, help='Skill 名称')
|
||
|
||
# remove-remote
|
||
remove_parser = subparsers.add_parser('remove-remote', help='移除远程 skill')
|
||
remove_parser.add_argument('--agent', required=True, help='Agent ID')
|
||
remove_parser.add_argument('--name', required=True, help='Skill 名称')
|
||
|
||
# import-official-hub
|
||
import_parser = subparsers.add_parser('import-official-hub', help='从官方库导入 skills')
|
||
import_parser.add_argument('--agents', default='', help='逗号分隔的 Agent IDs（可选）')
|
||
|
||
# check-updates
|
||
check_parser = subparsers.add_parser('check-updates', help='检查更新（未来功能）')
|
||
check_parser.add_argument('--interval', default='weekly',
|
||
help='检查间隔 (weekly/daily/monthly)')
|
||
|
||
args = parser.parse_args()
|
||
|
||
if not args.cmd:
|
||
parser.print_help()
|
||
return
|
||
|
||
if args.cmd == 'add-remote':
|
||
success = add_remote(args.agent, args.name, args.source, args.description)
|
||
sys.exit(0 if success else 1)
|
||
|
||
elif args.cmd == 'list-remote':
|
||
success = list_remote()
|
||
sys.exit(0 if success else 1)
|
||
|
||
elif args.cmd == 'update-remote':
|
||
success = update_remote(args.agent, args.name)
|
||
sys.exit(0 if success else 1)
|
||
|
||
elif args.cmd == 'remove-remote':
|
||
success = remove_remote(args.agent, args.name)
|
||
sys.exit(0 if success else 1)
|
||
|
||
elif args.cmd == 'import-official-hub':
|
||
agent_list = [a.strip() for a in args.agents.split(',') if a.strip()] if args.agents else []
|
||
success = import_official_hub(agent_list)
|
||
sys.exit(0 if success else 1)
|
||
|
||
elif args.cmd == 'check-updates':
|
||
print(f'⏳ 检查更新功能（间隔: {args.interval}）尚未实现')
|
||
print(f' 敬请期待...')
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|