Coverage for agentos/marketplace/registry.py: 0%

1""" 

2AgentOS Skill Marketplace — Registry。 

3 

4核心能力: 

5 - 本地注册表：~/.agentos/marketplace/installed.json 

6 - PyPI 发现：搜索 agentos-skill-* 前缀包 

7 - 安装/卸载/更新/搜索 

8 - 多格式兼容：agentos / openclaw / MCP / generic 

9 

10目录结构: 

11 ~/.agentos/marketplace/ 

12 installed.json # 已安装清单 

13 skills/ 

14 <name>/ # 每个 skill 一个目录 

15 manifest.yaml 

16 ... 

17""" 

18 

19from __future__ import annotations 

20 

21import json 

22import os 

23import shutil 

24import subprocess 

25import sys 

26import time 

27from dataclasses import dataclass, field 

28from pathlib import Path 

29from typing import Optional 

30 

31from agentos.marketplace.manifest import SkillManifest, SkillFormat 

32 

33MARKET_DIR = Path.home() / ".agentos" / "marketplace" 

34INSTALLED_JSON = MARKET_DIR / "installed.json" 

35SKILLS_DIR = MARKET_DIR / "skills" 

36PYPI_SKILL_PREFIX = "agentos-skill-" 

37 

38 

39class System: 

40 """简化后的 PyPI 搜索调用——用 pip 查询比 httpx 解析 JSON API 更可靠。""" 

41 

42 

43@dataclass 

44class SearchResult: 

45 """搜索结果。""" 

46 name: str 

47 version: str 

48 description: str 

49 source: str # pypi | github | local 

50 installable: bool = True 

51 pypi_package: str = "" 

52 skill_count: int = 0 

53 

54 

55@dataclass 

56class InstallResult: 

57 """安装结果。""" 

58 success: bool 

59 manifest: Optional[SkillManifest] = None 

60 error: str = "" 

61 pypi_package: str = "" 

62 install_type: str = "" # pypi_install | local_copy | git_clone 

63 dep_installed: list[str] = field(default_factory=list) 

64 

65 

66class SkillRegistry: 

67 """技能市场注册表。 

68 

69 支持三种安装源： 

70 1. PyPI 包（agentos-skill-* 前缀） 

71 2. 本地目录（含 manifest.yaml/json） 

72 3. GitHub 仓库（git clone + pip install） 

73 

74 每个 skill 安装后： 

75 - manifest 存入 installed.json 

76 - 源文件复制到 ~/.agentos/marketplace/skills/<name>/ 

77 - pip 依赖自动安装 

78 """ 

79 

80 def __init__(self): 

81 self._ensure_dirs() 

82 

83 # ── 搜索 ── 

84 

85 def search(self, query: str = "", max_results: int = 20) -> list[SearchResult]: 

86 """搜索技能市场。query 为空时返回热门。""" 

87 results: list[SearchResult] = [] 

88 

89 # 1. 搜索 PyPI（agentos-skill-*） 

90 try: 

91 r = subprocess.run( 

92 [sys.executable, "-m", "pip", "search", f"{PYPI_SKILL_PREFIX}{query}"] if query else 

93 [sys.executable, "-m", "pip", "search", PYPI_SKILL_PREFIX], 

94 capture_output=True, text=True, timeout=15, 

95 env={**os.environ, "PIP_DISABLE_PIP_VERSION_CHECK": "1"}, 

96 ) 

97 for line in r.stdout.split("\n"): 

98 line = line.strip() 

99 if line and PYPI_SKILL_PREFIX in line and not line.startswith("ERROR"): 

100 parts = line.split() 

101 pkg_name = parts[0] 

102 version = parts[1].lstrip("(").rstrip(")") if len(parts) > 1 else "?" 

103 desc = " ".join(parts[2:]) if len(parts) > 2 else "" 

104 skill_name = pkg_name.replace(PYPI_SKILL_PREFIX, "").replace("-", "_") 

105 # 去重 

106 if not any(r.name == skill_name for r in results): 

107 results.append(SearchResult( 

108 name=skill_name, 

109 version=version, 

110 description=desc, 

111 source="pypi", 

112 pypi_package=pkg_name, 

113 )) 

114 except Exception: 

115 pass 

116 

117 # 2. 如果 pip search 不可用，尝试直接查 PyPI JSON API 

118 if not results and query: 

119 try: 

120 import urllib.request 

121 url = f"https://pypi.org/pypi/{PYPI_SKILL_PREFIX}{query}/json" 

122 req = urllib.request.Request(url, headers={"User-Agent": "AgentOS-Marketplace/1.0"}) 

123 with urllib.request.urlopen(req, timeout=8) as resp: 

124 data = json.loads(resp.read()) 

125 info = data.get("info", {}) 

126 skill_name = query.replace("-", "_") 

127 results.append(SearchResult( 

128 name=skill_name, 

129 version=info.get("version", "?"), 

130 description=info.get("summary", ""), 

131 source="pypi", 

132 pypi_package=f"{PYPI_SKILL_PREFIX}{query}", 

133 )) 

134 except Exception: 

135 pass 

136 

137 # 3. 限制结果数 

138 return results[:max_results] 

139 

140 def list_installed(self) -> list[SkillManifest]: 

141 """列出所有已安装 skill。""" 

142 data = self._load_installed() 

143 manifests = [] 

144 for entry in data.get("skills", []): 

145 m = SkillManifest.from_dict(entry) 

146 if m.name: 

147 manifests.append(m) 

148 return sorted(manifests, key=lambda m: m.name) 

149 

150 def get_installed(self, name: str) -> Optional[SkillManifest]: 

151 """获取已安装 skill 的 manifest。""" 

152 for m in self.list_installed(): 

153 if m.name == name: 

154 return m 

155 return None 

156 

157 # ── 安装 ── 

158 

159 def install(self, name_or_path: str) -> InstallResult: 

160 """安装一个 skill。 

161 

162 自动判断安装源： 

163 1. 本地目录（包含 manifest.yaml/json 时复制安装） 

164 2. Git URL（含 github.com 时 git clone） 

165 3. PyPI 包（pip install agentos-skill-<name>） 

166 """ 

167 target = name_or_path.strip() 

168 

169 # ─ 源 1: 本地目录 ─ 

170 local_path = Path(name_or_path) 

171 if local_path.is_dir(): 

172 manifest_file = local_path / "skill.yaml" 

173 if not manifest_file.exists(): 

174 manifest_file = local_path / "skill.json" 

175 if not manifest_file.exists(): 

176 manifest_file = local_path / "manifest.yaml" 

177 if not manifest_file.exists(): 

178 manifest_file = local_path / "manifest.json" 

179 if manifest_file.exists(): 

180 return self._install_local(local_path, manifest_file) 

181 return InstallResult(success=False, error=f"No manifest (skill.yaml/json) found in {local_path}") 

182 

183 # ─ 源 2: GitHub URL ─ 

184 if "github.com" in name_or_path: 

185 return self._install_github(name_or_path) 

186 

187 # ─ 源 3: PyPI 包 ─ 

188 return self._install_pypi(target) 

189 

190 def uninstall(self, name: str) -> bool: 

191 """卸载一个 skill。""" 

192 existing = self.get_installed(name) 

193 if not existing: 

194 return False 

195 

196 # 删除 skill 目录 

197 skill_dir = SKILLS_DIR / name 

198 if skill_dir.exists(): 

199 shutil.rmtree(skill_dir) 

200 

201 # 更新 installed.json 

202 data = self._load_installed() 

203 data["skills"] = [s for s in data.get("skills", []) if s.get("name") != name] 

204 self._save_installed(data) 

205 return True 

206 

207 def update(self, name: str) -> InstallResult: 

208 """更新一个 skill 到最新版。""" 

209 existing = self.get_installed(name) 

210 if not existing: 

211 return InstallResult(success=False, error=f"Skill '{name}' not installed.") 

212 

213 # 先卸载再重新安装 

214 old_source = existing.source 

215 self.uninstall(name) 

216 

217 if old_source == "pypi": 

218 return self._install_pypi(name) 

219 elif old_source == "github": 

220 return self._install_github(existing.repository or f"https://github.com/{name}") 

221 elif old_source == "local": 

222 return self._install_local(Path(existing.install_path), Path(existing.install_path) / "skill.yaml") 

223 

224 return InstallResult(success=False, error=f"Unknown source: {old_source}") 

225 

226 def register(self, manifest: SkillManifest, force: bool = False) -> Optional[InstallResult]: 

227 """公开注册接口: 直接将 SkillManifest 写入 registry (不复制文件)。 

228 

229 用于 importer.import_skill() / import_all() 等场景。 

230 与 install() 的区别: install 会复制/下载源文件，register 只写索引。 

231 """ 

232 existing = self.get_installed(manifest.name) 

233 if existing and not force: 

234 return InstallResult( 

235 success=False, 

236 error=f"Skill '{manifest.name}' already installed. Use force=True to overwrite.", 

237 ) 

238 if existing and force: 

239 self.uninstall(manifest.name) 

240 

241 self._register_manifest(manifest) 

242 return InstallResult(success=True, manifest=manifest) 

243 

244 def stats(self) -> dict: 

245 """市场统计。""" 

246 installed = self.list_installed() 

247 by_format = {} 

248 for m in installed: 

249 fmt = m.format.value 

250 by_format[fmt] = by_format.get(fmt, 0) + 1 

251 return { 

252 "total": len(installed), 

253 "by_format": by_format, 

254 "market_dir": str(MARKET_DIR), 

255 } 

256 

257 def _check_duplicate(self, name: str) -> Optional[InstallResult]: 

258 """如果 skill 已安装，返回失败结果；否则返回 None。""" 

259 existing = self.get_installed(name) 

260 if existing: 

261 return InstallResult( 

262 success=False, 

263 error=f"Skill '{name}' already installed (v{existing.version}). Use 'marketplace update {name}' to upgrade.", 

264 ) 

265 return None 

266 

267 def _install_pypi(self, name: str) -> InstallResult: 

268 """从 PyPI 安装 agentos-skill-<name>。""" 

269 pkg = f"{PYPI_SKILL_PREFIX}{name}" 

270 try: 

271 r = subprocess.run( 

272 [sys.executable, "-m", "pip", "install", pkg, "--quiet", "--disable-pip-version-check"], 

273 capture_output=True, text=True, timeout=120, 

274 ) 

275 if r.returncode != 0: 

276 return InstallResult(success=False, error=f"pip install failed: {r.stderr[:200]}") 

277 except subprocess.TimeoutExpired: 

278 return InstallResult(success=False, error="pip install timed out") 

279 

280 # 查找安装后的包位置，读取 manifest 

281 manifest = self._find_package_manifest(pkg) 

282 if not manifest: 

283 # 没有 manifest 的 PyPI 包也注册为 generic skill 

284 manifest = SkillManifest( 

285 name=name, 

286 version="?", 

287 description=f"PyPI package: {pkg}", 

288 format=SkillFormat.GENERIC, 

289 source="pypi", 

290 ) 

291 manifest.source = "pypi" 

292 

293 # 复制到 skills 目录 

294 self._copy_to_skills(name, manifest) 

295 

296 # 注册 

297 self._register_manifest(manifest) 

298 return InstallResult( 

299 success=True, 

300 manifest=manifest, 

301 install_type="pypi_install", 

302 pypi_package=pkg, 

303 ) 

304 

305 def _install_local(self, local_path: Path, manifest_file: Path) -> InstallResult: 

306 """从本地目录安装。""" 

307 raw = manifest_file.read_text(encoding="utf-8") 

308 if manifest_file.suffix in (".yaml", ".yml"): 

309 import yaml 

310 data = yaml.safe_load(raw) or {} 

311 else: 

312 data = json.loads(raw) 

313 

314 manifest = SkillManifest.from_dict(data, source="local", install_path=str(local_path)) 

315 name = manifest.name or local_path.name 

316 

317 dup = self._check_duplicate(name) 

318 if dup: 

319 return dup 

320 

321 # 安装依赖 

322 deps = self._install_deps(manifest) 

323 

324 # 复制到 skills 目录 

325 dest = SKILLS_DIR / name 

326 if dest.exists(): 

327 shutil.rmtree(dest) 

328 shutil.copytree(local_path, dest) 

329 

330 manifest.install_path = str(dest) 

331 manifest.source = "local" 

332 self._register_manifest(manifest) 

333 

334 return InstallResult( 

335 success=True, 

336 manifest=manifest, 

337 install_type="local_copy", 

338 dep_installed=deps, 

339 ) 

340 

341 def _install_github(self, url: str) -> InstallResult: 

342 """从 GitHub 克隆安装。""" 

343 name = url.rstrip("/").split("/")[-1].replace(".git", "") 

344 dest = SKILLS_DIR / name 

345 

346 if dest.exists(): 

347 shutil.rmtree(dest) 

348 

349 try: 

350 r = subprocess.run( 

351 ["git", "clone", "--depth=1", url, str(dest)], 

352 capture_output=True, text=True, timeout=60, 

353 ) 

354 if r.returncode != 0: 

355 return InstallResult(success=False, error=f"git clone failed: {r.stderr[:200]}") 

356 except FileNotFoundError: 

357 return InstallResult(success=False, error="git not found. Install git first.") 

358 except subprocess.TimeoutExpired: 

359 return InstallResult(success=False, error="git clone timed out") 

360 

361 # 查找 manifest 文件 

362 manifest_file = None 

363 for fname in ("skill.yaml", "skill.json", "manifest.yaml", "manifest.json"): 

364 candidate = dest / fname 

365 if candidate.exists(): 

366 manifest_file = candidate 

367 break 

368 

369 if manifest_file: 

370 raw = manifest_file.read_text(encoding="utf-8") 

371 if manifest_file.suffix in (".yaml", ".yml"): 

372 import yaml 

373 data = yaml.safe_load(raw) 

374 else: 

375 data = json.loads(raw) 

376 manifest = SkillManifest.from_dict(data, source="github", install_path=str(dest)) 

377 else: 

378 manifest = SkillManifest( 

379 name=name, 

380 version="0.1.0", 

381 description=f"GitHub skill: {url}", 

382 format=SkillFormat.GENERIC, 

383 source="github", 

384 install_path=str(dest), 

385 repository=url, 

386 ) 

387 

388 manifest.source = "github" 

389 manifest.repository = url 

390 if not manifest.name: 

391 manifest.name = name 

392 

393 deps = self._install_deps(manifest) 

394 self._register_manifest(manifest) 

395 

396 return InstallResult( 

397 success=True, 

398 manifest=manifest, 

399 install_type="git_clone", 

400 dep_installed=deps, 

401 ) 

402 

403 def _install_deps(self, manifest: SkillManifest) -> list[str]: 

404 """安装 skill 的 pip 依赖，返回成功安装的包名列表。""" 

405 installed = [] 

406 for dep in manifest.dependencies: 

407 try: 

408 subprocess.run( 

409 [sys.executable, "-m", "pip", "install", dep, "--quiet", "--disable-pip-version-check"], 

410 capture_output=True, timeout=60, 

411 ) 

412 installed.append(dep) 

413 except Exception: 

414 pass 

415 return installed 

416 

417 def _find_package_manifest(self, pkg_name: str) -> Optional[SkillManifest]: 

418 """从已安装的 PyPI 包中查找 manifest。""" 

419 try: 

420 r = subprocess.run( 

421 [sys.executable, "-m", "pip", "show", "-f", pkg_name], 

422 capture_output=True, text=True, timeout=10, 

423 ) 

424 if r.returncode != 0: 

425 return None 

426 

427 # 找 Location 行 

428 location = "" 

429 for line in r.stdout.split("\n"): 

430 if line.startswith("Location:"): 

431 location = line.split(":", 1)[1].strip() 

432 break 

433 

434 if not location: 

435 return None 

436 

437 # 尝试常见 manifest 路径 

438 pkg_name_clean = pkg_name.replace("-", "_") 

439 candidates = [ 

440 Path(location) / pkg_name_clean / "skill.yaml", 

441 Path(location) / pkg_name_clean / "skill.json", 

442 Path(location) / pkg_name_clean / "manifest.yaml", 

443 Path(location) / pkg_name_clean / "manifest.json", 

444 ] 

445 for p in candidates: 

446 if p.exists(): 

447 raw = p.read_text(encoding="utf-8") 

448 if p.suffix in (".yaml", ".yml"): 

449 import yaml 

450 data = yaml.safe_load(raw) 

451 else: 

452 data = json.loads(raw) 

453 return SkillManifest.from_dict(data, source="pypi") 

454 except Exception: 

455 pass 

456 return None 

457 

458 def _copy_to_skills(self, name: str, manifest: SkillManifest): 

459 """确保 skill 源文件在 skills 目录有一份。""" 

460 dest = SKILLS_DIR / name 

461 dest.mkdir(parents=True, exist_ok=True) 

462 manifest_path = dest / "manifest.yaml" 

463 import yaml 

464 manifest_path.write_text( 

465 yaml.dump(manifest.to_dict(), allow_unicode=True, default_flow_style=False, sort_keys=False), 

466 encoding="utf-8", 

467 ) 

468 

469 def _register_manifest(self, manifest: SkillManifest): 

470 """将 manifest 注册到 installed.json。""" 

471 data = self._load_installed() 

472 # 去重 

473 data["skills"] = [s for s in data.get("skills", []) if s.get("name") != manifest.name] 

474 entry = manifest.to_dict() 

475 entry["installed_at"] = time.time() 

476 data["skills"].append(entry) 

477 self._save_installed(data) 

478 

479 def _load_installed(self) -> dict: 

480 MARKET_DIR.mkdir(parents=True, exist_ok=True) 

481 if not INSTALLED_JSON.exists(): 

482 return {"version": "1.0", "skills": []} 

483 try: 

484 return json.loads(INSTALLED_JSON.read_text(encoding="utf-8")) 

485 except (json.JSONDecodeError, Exception): 

486 return {"version": "1.0", "skills": []} 

487 

488 def _save_installed(self, data: dict): 

489 MARKET_DIR.mkdir(parents=True, exist_ok=True) 

490 INSTALLED_JSON.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8") 

491 

492 def _ensure_dirs(self): 

493 MARKET_DIR.mkdir(parents=True, exist_ok=True) 

494 SKILLS_DIR.mkdir(parents=True, exist_ok=True)