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

1""" # noqa: E501

2AgentOS Skill Marketplace — Registry。

4核心能力:

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

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

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

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

10目录结构:

11 ~/.agentos/marketplace/

12 installed.json # 已安装清单

13 skills/

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

15 manifest.yaml

16 ...

17"""

19from __future__ import annotations

21import json

22import os

23import shutil

24import subprocess

25import sys

26import time

27from dataclasses import dataclass, field

28from pathlib import Path

30from agentos.marketplace.manifest import SkillFormat, SkillManifest

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

33INSTALLED_JSON = MARKET_DIR / "installed.json"

34SKILLS_DIR = MARKET_DIR / "skills"

35PYPI_SKILL_PREFIX = "agentos-skill-"

38class System:

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

42@dataclass

43class SearchResult:

44 """搜索结果。"""

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

55@dataclass

56class InstallResult:

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

59 success: bool

60 manifest: SkillManifest | None = None

61 error: str = ""

62 pypi_package: str = ""

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

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

67class SkillRegistry:

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

70 支持三种安装源：

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

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

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

75 每个 skill 安装后：

76 - manifest 存入 installed.json

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

78 - pip 依赖自动安装

79 """

81 def __init__(self):

82 self._ensure_dirs()

84 # ── 搜索 ──

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

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

88 results: list[SearchResult] = []

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

91 try:

92 r = subprocess.run(

93 (

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

95 if query

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

97 ),

98 capture_output=True,

99 text=True,

100 timeout=15,

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

102 )

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

104 line = line.strip()

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

106 parts = line.split()

107 pkg_name = parts[0]

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

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

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

111 # 去重

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

113 results.append(

114 SearchResult(

115 name=skill_name,

116 version=version,

117 description=desc,

118 source="pypi",

119 pypi_package=pkg_name,

120 )

121 )

122 except Exception:

123 pass

124

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

126 if not results and query:

127 try:

128 import urllib.request

129

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

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

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

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

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

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

136 results.append(

137 SearchResult(

138 name=skill_name,

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

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

141 source="pypi",

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

143 )

144 )

145 except Exception:

146 pass

147

148 # 3. 限制结果数

149 return results[:max_results]

150

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

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

153 data = self._load_installed()

154 manifests = []

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

156 m = SkillManifest.from_dict(entry)

157 if m.name:

158 manifests.append(m)

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

160

161 def get_installed(self, name: str) -> SkillManifest | None:

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

163 for m in self.list_installed():

164 if m.name == name:

165 return m

166 return None

167

168 # ── 安装 ──

169

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

171 """安装一个 skill。

172

173 自动判断安装源：

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

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

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

177 """

178 target = name_or_path.strip()

179

180 # ─ 源 1: 本地目录 ─

181 local_path = Path(name_or_path)

182 if local_path.is_dir():

183 manifest_file = local_path / "skill.yaml"

184 if not manifest_file.exists():

185 manifest_file = local_path / "skill.json"

186 if not manifest_file.exists():

187 manifest_file = local_path / "manifest.yaml"

188 if not manifest_file.exists():

189 manifest_file = local_path / "manifest.json"

190 if manifest_file.exists():

191 return self._install_local(local_path, manifest_file)

192 return InstallResult(

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

194 )

195

196 # ─ 源 2: GitHub URL ─

197 if "github.com" in name_or_path:

198 return self._install_github(name_or_path)

199

200 # ─ 源 3: PyPI 包 ─

201 return self._install_pypi(target)

202

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

204 """卸载一个 skill。"""

205 existing = self.get_installed(name)

206 if not existing:

207 return False

208

209 # 删除 skill 目录

210 skill_dir = SKILLS_DIR / name

211 if skill_dir.exists():

212 shutil.rmtree(skill_dir)

213

214 # 更新 installed.json

215 data = self._load_installed()

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

217 self._save_installed(data)

218 return True

219

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

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

222 existing = self.get_installed(name)

223 if not existing:

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

225

226 # 先卸载再重新安装

227 old_source = existing.source

228 self.uninstall(name)

229

230 if old_source == "pypi":

231 return self._install_pypi(name)

232 elif old_source == "github":

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

234 elif old_source == "local":

235 return self._install_local(

236 Path(existing.install_path), Path(existing.install_path) / "skill.yaml"

237 )

238

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

240

241 def register(self, manifest: SkillManifest, force: bool = False) -> InstallResult | None:

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

243

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

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

246 """

247 existing = self.get_installed(manifest.name)

248 if existing and not force:

249 return InstallResult(

250 success=False,

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

252 )

253 if existing and force:

254 self.uninstall(manifest.name)

255

256 self._register_manifest(manifest)

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

258

259 def stats(self) -> dict:

260 """市场统计。"""

261 installed = self.list_installed()

262 by_format = {}

263 for m in installed:

264 fmt = m.format.value

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

266 return {

267 "total": len(installed),

268 "by_format": by_format,

269 "market_dir": str(MARKET_DIR),

270 }

271

272 def _check_duplicate(self, name: str) -> InstallResult | None:

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

274 existing = self.get_installed(name)

275 if existing:

276 return InstallResult(

277 success=False,

278 error=f"Skill '{name}' already installed (v{existing.version}). Use 'marketplace update {name}' to upgrade.", # noqa: E501

279 )

280 return None

281

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

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

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

285 try:

286 r = subprocess.run(

287 [

288 sys.executable,

289 "-m",

290 "pip",

291 "install",

292 pkg,

293 "--quiet",

294 "--disable-pip-version-check",

295 ],

296 capture_output=True,

297 text=True,

298 timeout=120,

299 )

300 if r.returncode != 0:

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

302 except subprocess.TimeoutExpired:

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

304

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

306 manifest = self._find_package_manifest(pkg)

307 if not manifest:

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

309 manifest = SkillManifest(

310 name=name,

311 version="?",

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

313 format=SkillFormat.GENERIC,

314 source="pypi",

315 )

316 manifest.source = "pypi"

317

318 # 复制到 skills 目录

319 self._copy_to_skills(name, manifest)

320

321 # 注册

322 self._register_manifest(manifest)

323 return InstallResult(

324 success=True,

325 manifest=manifest,

326 install_type="pypi_install",

327 pypi_package=pkg,

328 )

329

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

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

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

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

334 import yaml

335

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

337 else:

338 data = json.loads(raw)

339

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

341 name = manifest.name or local_path.name

342

343 dup = self._check_duplicate(name)

344 if dup:

345 return dup

346

347 # 安装依赖

348 deps = self._install_deps(manifest)

349

350 # 复制到 skills 目录

351 dest = SKILLS_DIR / name

352 if dest.exists():

353 shutil.rmtree(dest)

354 shutil.copytree(local_path, dest)

355

356 manifest.install_path = str(dest)

357 manifest.source = "local"

358 self._register_manifest(manifest)

359

360 return InstallResult(

361 success=True,

362 manifest=manifest,

363 install_type="local_copy",

364 dep_installed=deps,

365 )

366

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

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

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

370 dest = SKILLS_DIR / name

371

372 if dest.exists():

373 shutil.rmtree(dest)

374

375 try:

376 r = subprocess.run(

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

378 capture_output=True,

379 text=True,

380 timeout=60,

381 )

382 if r.returncode != 0:

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

384 except FileNotFoundError:

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

386 except subprocess.TimeoutExpired:

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

388

389 # 查找 manifest 文件

390 manifest_file = None

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

392 candidate = dest / fname

393 if candidate.exists():

394 manifest_file = candidate

395 break

396

397 if manifest_file:

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

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

400 import yaml

401

402 data = yaml.safe_load(raw)

403 else:

404 data = json.loads(raw)

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

406 else:

407 manifest = SkillManifest(

408 name=name,

409 version="0.1.0",

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

411 format=SkillFormat.GENERIC,

412 source="github",

413 install_path=str(dest),

414 repository=url,

415 )

416

417 manifest.source = "github"

418 manifest.repository = url

419 if not manifest.name:

420 manifest.name = name

421

422 deps = self._install_deps(manifest)

423 self._register_manifest(manifest)

424

425 return InstallResult(

426 success=True,

427 manifest=manifest,

428 install_type="git_clone",

429 dep_installed=deps,

430 )

431

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

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

434 installed = []

435 for dep in manifest.dependencies:

436 try:

437 subprocess.run(

438 [

439 sys.executable,

440 "-m",

441 "pip",

442 "install",

443 dep,

444 "--quiet",

445 "--disable-pip-version-check",

446 ],

447 capture_output=True,

448 timeout=60,

449 )

450 installed.append(dep)

451 except Exception:

452 pass

453 return installed

454

455 def _find_package_manifest(self, pkg_name: str) -> SkillManifest | None:

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

457 try:

458 r = subprocess.run(

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

460 capture_output=True,

461 text=True,

462 timeout=10,

463 )

464 if r.returncode != 0:

465 return None

466

467 # 找 Location 行

468 location = ""

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

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

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

472 break

473

474 if not location:

475 return None

476

477 # 尝试常见 manifest 路径

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

479 candidates = [

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

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

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

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

484 ]

485 for p in candidates:

486 if p.exists():

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

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

489 import yaml

490

491 data = yaml.safe_load(raw)

492 else:

493 data = json.loads(raw)

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

495 except Exception:

496 pass

497 return None

498

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

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

501 dest = SKILLS_DIR / name

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

503 manifest_path = dest / "manifest.yaml"

504 import yaml

505

506 manifest_path.write_text(

507 yaml.dump(

508 manifest.to_dict(), allow_unicode=True, default_flow_style=False, sort_keys=False

509 ),

510 encoding="utf-8",

511 )

512

513 def _register_manifest(self, manifest: SkillManifest):

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

515 data = self._load_installed()

516 # 去重

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

518 entry = manifest.to_dict()

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

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

521 self._save_installed(data)

522

523 def _load_installed(self) -> dict:

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

525 if not INSTALLED_JSON.exists():

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

527 try:

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

529 except (json.JSONDecodeError, Exception):

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

531

532 def _save_installed(self, data: dict):

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

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

535

536 def _ensure_dirs(self):

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

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