Coverage for agentos/marketplace/registry.py: 0%
270 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-05 20:52 +0800
1"""
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 tempfile
27import time
28from dataclasses import dataclass, field
29from pathlib import Path
30from typing import Optional
32from agentos.marketplace.manifest import SkillManifest, SkillFormat, ToolDef
34MARKET_DIR = Path.home() / ".agentos" / "marketplace"
35INSTALLED_JSON = MARKET_DIR / "installed.json"
36SKILLS_DIR = MARKET_DIR / "skills"
37PYPI_SKILL_PREFIX = "agentos-skill-"
40class System:
41 """简化后的 PyPI 搜索调用——用 pip 查询比 httpx 解析 JSON API 更可靠。"""
44@dataclass
45class SearchResult:
46 """搜索结果。"""
47 name: str
48 version: str
49 description: str
50 source: str # pypi | github | local
51 installable: bool = True
52 pypi_package: str = ""
53 skill_count: int = 0
56@dataclass
57class InstallResult:
58 """安装结果。"""
59 success: bool
60 manifest: Optional[SkillManifest] = 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 [sys.executable, "-m", "pip", "search", f"{PYPI_SKILL_PREFIX}{query}"] if query else
94 [sys.executable, "-m", "pip", "search", PYPI_SKILL_PREFIX],
95 capture_output=True, text=True, timeout=15,
96 env={**os.environ, "PIP_DISABLE_PIP_VERSION_CHECK": "1"},
97 )
98 for line in r.stdout.split("\n"):
99 line = line.strip()
100 if line and PYPI_SKILL_PREFIX in line and not line.startswith("ERROR"):
101 parts = line.split()
102 pkg_name = parts[0]
103 version = parts[1].lstrip("(").rstrip(")") if len(parts) > 1 else "?"
104 desc = " ".join(parts[2:]) if len(parts) > 2 else ""
105 skill_name = pkg_name.replace(PYPI_SKILL_PREFIX, "").replace("-", "_")
106 # 去重
107 if not any(r.name == skill_name for r in results):
108 results.append(SearchResult(
109 name=skill_name,
110 version=version,
111 description=desc,
112 source="pypi",
113 pypi_package=pkg_name,
114 ))
115 except Exception:
116 pass
118 # 2. 如果 pip search 不可用，尝试直接查 PyPI JSON API
119 if not results and query:
120 try:
121 import urllib.request
122 url = f"https://pypi.org/pypi/{PYPI_SKILL_PREFIX}{query}/json"
123 req = urllib.request.Request(url, headers={"User-Agent": "AgentOS-Marketplace/1.0"})
124 with urllib.request.urlopen(req, timeout=8) as resp:
125 data = json.loads(resp.read())
126 info = data.get("info", {})
127 skill_name = query.replace("-", "_")
128 results.append(SearchResult(
129 name=skill_name,
130 version=info.get("version", "?"),
131 description=info.get("summary", ""),
132 source="pypi",
133 pypi_package=f"{PYPI_SKILL_PREFIX}{query}",
134 ))
135 except Exception:
136 pass
138 # 3. 限制结果数
139 return results[:max_results]
141 def list_installed(self) -> list[SkillManifest]:
142 """列出所有已安装 skill。"""
143 data = self._load_installed()
144 manifests = []
145 for entry in data.get("skills", []):
146 m = SkillManifest.from_dict(entry)
147 if m.name:
148 manifests.append(m)
149 return sorted(manifests, key=lambda m: m.name)
151 def get_installed(self, name: str) -> Optional[SkillManifest]:
152 """获取已安装 skill 的 manifest。"""
153 for m in self.list_installed():
154 if m.name == name:
155 return m
156 return None
158 # ── 安装 ──
160 def install(self, name_or_path: str) -> InstallResult:
161 """安装一个 skill。
163 自动判断安装源：
164 1. 本地目录（包含 manifest.yaml/json 时复制安装）
165 2. Git URL（含 github.com 时 git clone）
166 3. PyPI 包（pip install agentos-skill-<name>）
167 """
168 target = name_or_path.strip()
170 # ─ 源 1: 本地目录 ─
171 local_path = Path(name_or_path)
172 if local_path.is_dir():
173 manifest_file = local_path / "skill.yaml"
174 if not manifest_file.exists():
175 manifest_file = local_path / "skill.json"
176 if not manifest_file.exists():
177 manifest_file = local_path / "manifest.yaml"
178 if not manifest_file.exists():
179 manifest_file = local_path / "manifest.json"
180 if manifest_file.exists():
181 return self._install_local(local_path, manifest_file)
182 return InstallResult(success=False, error=f"No manifest (skill.yaml/json) found in {local_path}")
184 # ─ 源 2: GitHub URL ─
185 if "github.com" in name_or_path:
186 return self._install_github(name_or_path)
188 # ─ 源 3: PyPI 包 ─
189 return self._install_pypi(target)
191 def uninstall(self, name: str) -> bool:
192 """卸载一个 skill。"""
193 existing = self.get_installed(name)
194 if not existing:
195 return False
197 # 删除 skill 目录
198 skill_dir = SKILLS_DIR / name
199 if skill_dir.exists():
200 shutil.rmtree(skill_dir)
202 # 更新 installed.json
203 data = self._load_installed()
204 data["skills"] = [s for s in data.get("skills", []) if s.get("name") != name]
205 self._save_installed(data)
206 return True
208 def update(self, name: str) -> InstallResult:
209 """更新一个 skill 到最新版。"""
210 existing = self.get_installed(name)
211 if not existing:
212 return InstallResult(success=False, error=f"Skill '{name}' not installed.")
214 # 先卸载再重新安装
215 old_source = existing.source
216 self.uninstall(name)
218 if old_source == "pypi":
219 return self._install_pypi(name)
220 elif old_source == "github":
221 return self._install_github(existing.repository or f"https://github.com/{name}")
222 elif old_source == "local":
223 return self._install_local(Path(existing.install_path), Path(existing.install_path) / "skill.yaml")
225 return InstallResult(success=False, error=f"Unknown source: {old_source}")
227 def register(self, manifest: SkillManifest, force: bool = False) -> Optional[InstallResult]:
228 """公开注册接口: 直接将 SkillManifest 写入 registry (不复制文件)。
230 用于 importer.import_skill() / import_all() 等场景。
231 与 install() 的区别: install 会复制/下载源文件，register 只写索引。
232 """
233 existing = self.get_installed(manifest.name)
234 if existing and not force:
235 return InstallResult(
236 success=False,
237 error=f"Skill '{manifest.name}' already installed. Use force=True to overwrite.",
238 )
239 if existing and force:
240 self.uninstall(manifest.name)
242 self._register_manifest(manifest)
243 return InstallResult(success=True, manifest=manifest)
245 def stats(self) -> dict:
246 """市场统计。"""
247 installed = self.list_installed()
248 by_format = {}
249 for m in installed:
250 fmt = m.format.value
251 by_format[fmt] = by_format.get(fmt, 0) + 1
252 return {
253 "total": len(installed),
254 "by_format": by_format,
255 "market_dir": str(MARKET_DIR),
256 }
258 def _check_duplicate(self, name: str) -> Optional[InstallResult]:
259 """如果 skill 已安装，返回失败结果；否则返回 None。"""
260 existing = self.get_installed(name)
261 if existing:
262 return InstallResult(
263 success=False,
264 error=f"Skill '{name}' already installed (v{existing.version}). Use 'marketplace update {name}' to upgrade.",
265 )
266 return None
268 def _install_pypi(self, name: str) -> InstallResult:
269 """从 PyPI 安装 agentos-skill-<name>。"""
270 pkg = f"{PYPI_SKILL_PREFIX}{name}"
271 try:
272 r = subprocess.run(
273 [sys.executable, "-m", "pip", "install", pkg, "--quiet", "--disable-pip-version-check"],
274 capture_output=True, text=True, timeout=120,
275 )
276 if r.returncode != 0:
277 return InstallResult(success=False, error=f"pip install failed: {r.stderr[:200]}")
278 except subprocess.TimeoutExpired:
279 return InstallResult(success=False, error="pip install timed out")
281 # 查找安装后的包位置，读取 manifest
282 manifest = self._find_package_manifest(pkg)
283 if not manifest:
284 # 没有 manifest 的 PyPI 包也注册为 generic skill
285 manifest = SkillManifest(
286 name=name,
287 version="?",
288 description=f"PyPI package: {pkg}",
289 format=SkillFormat.GENERIC,
290 source="pypi",
291 )
292 manifest.source = "pypi"
294 # 复制到 skills 目录
295 self._copy_to_skills(name, manifest)
297 # 注册
298 self._register_manifest(manifest)
299 return InstallResult(
300 success=True,
301 manifest=manifest,
302 install_type="pypi_install",
303 pypi_package=pkg,
304 )
306 def _install_local(self, local_path: Path, manifest_file: Path) -> InstallResult:
307 """从本地目录安装。"""
308 raw = manifest_file.read_text(encoding="utf-8")
309 if manifest_file.suffix in (".yaml", ".yml"):
310 import yaml
311 data = yaml.safe_load(raw) or {}
312 else:
313 data = json.loads(raw)
315 manifest = SkillManifest.from_dict(data, source="local", install_path=str(local_path))
316 name = manifest.name or local_path.name
318 dup = self._check_duplicate(name)
319 if dup:
320 return dup
322 # 安装依赖
323 deps = self._install_deps(manifest)
325 # 复制到 skills 目录
326 dest = SKILLS_DIR / name
327 if dest.exists():
328 shutil.rmtree(dest)
329 shutil.copytree(local_path, dest)
331 manifest.install_path = str(dest)
332 manifest.source = "local"
333 self._register_manifest(manifest)
335 return InstallResult(
336 success=True,
337 manifest=manifest,
338 install_type="local_copy",
339 dep_installed=deps,
340 )
342 def _install_github(self, url: str) -> InstallResult:
343 """从 GitHub 克隆安装。"""
344 name = url.rstrip("/").split("/")[-1].replace(".git", "")
345 dest = SKILLS_DIR / name
347 if dest.exists():
348 shutil.rmtree(dest)
350 try:
351 r = subprocess.run(
352 ["git", "clone", "--depth=1", url, str(dest)],
353 capture_output=True, text=True, timeout=60,
354 )
355 if r.returncode != 0:
356 return InstallResult(success=False, error=f"git clone failed: {r.stderr[:200]}")
357 except FileNotFoundError:
358 return InstallResult(success=False, error="git not found. Install git first.")
359 except subprocess.TimeoutExpired:
360 return InstallResult(success=False, error="git clone timed out")
362 # 查找 manifest 文件
363 manifest_file = None
364 for fname in ("skill.yaml", "skill.json", "manifest.yaml", "manifest.json"):
365 candidate = dest / fname
366 if candidate.exists():
367 manifest_file = candidate
368 break
370 if manifest_file:
371 raw = manifest_file.read_text(encoding="utf-8")
372 if manifest_file.suffix in (".yaml", ".yml"):
373 import yaml
374 data = yaml.safe_load(raw)
375 else:
376 data = json.loads(raw)
377 manifest = SkillManifest.from_dict(data, source="github", install_path=str(dest))
378 else:
379 manifest = SkillManifest(
380 name=name,
381 version="0.1.0",
382 description=f"GitHub skill: {url}",
383 format=SkillFormat.GENERIC,
384 source="github",
385 install_path=str(dest),
386 repository=url,
387 )
389 manifest.source = "github"
390 manifest.repository = url
391 if not manifest.name:
392 manifest.name = name
394 deps = self._install_deps(manifest)
395 self._register_manifest(manifest)
397 return InstallResult(
398 success=True,
399 manifest=manifest,
400 install_type="git_clone",
401 dep_installed=deps,
402 )
404 def _install_deps(self, manifest: SkillManifest) -> list[str]:
405 """安装 skill 的 pip 依赖，返回成功安装的包名列表。"""
406 installed = []
407 for dep in manifest.dependencies:
408 try:
409 subprocess.run(
410 [sys.executable, "-m", "pip", "install", dep, "--quiet", "--disable-pip-version-check"],
411 capture_output=True, timeout=60,
412 )
413 installed.append(dep)
414 except Exception:
415 pass
416 return installed
418 def _find_package_manifest(self, pkg_name: str) -> Optional[SkillManifest]:
419 """从已安装的 PyPI 包中查找 manifest。"""
420 try:
421 r = subprocess.run(
422 [sys.executable, "-m", "pip", "show", "-f", pkg_name],
423 capture_output=True, text=True, timeout=10,
424 )
425 if r.returncode != 0:
426 return None
428 # 找 Location 行
429 location = ""
430 for line in r.stdout.split("\n"):
431 if line.startswith("Location:"):
432 location = line.split(":", 1)[1].strip()
433 break
435 if not location:
436 return None
438 # 尝试常见 manifest 路径
439 pkg_name_clean = pkg_name.replace("-", "_")
440 candidates = [
441 Path(location) / pkg_name_clean / "skill.yaml",
442 Path(location) / pkg_name_clean / "skill.json",
443 Path(location) / pkg_name_clean / "manifest.yaml",
444 Path(location) / pkg_name_clean / "manifest.json",
445 ]
446 for p in candidates:
447 if p.exists():
448 raw = p.read_text(encoding="utf-8")
449 if p.suffix in (".yaml", ".yml"):
450 import yaml
451 data = yaml.safe_load(raw)
452 else:
453 data = json.loads(raw)
454 return SkillManifest.from_dict(data, source="pypi")
455 except Exception:
456 pass
457 return None
459 def _copy_to_skills(self, name: str, manifest: SkillManifest):
460 """确保 skill 源文件在 skills 目录有一份。"""
461 dest = SKILLS_DIR / name
462 dest.mkdir(parents=True, exist_ok=True)
463 manifest_path = dest / "manifest.yaml"
464 import yaml
465 manifest_path.write_text(
466 yaml.dump(manifest.to_dict(), allow_unicode=True, default_flow_style=False, sort_keys=False),
467 encoding="utf-8",
468 )
470 def _register_manifest(self, manifest: SkillManifest):
471 """将 manifest 注册到 installed.json。"""
472 data = self._load_installed()
473 # 去重
474 data["skills"] = [s for s in data.get("skills", []) if s.get("name") != manifest.name]
475 entry = manifest.to_dict()
476 entry["installed_at"] = time.time()
477 data["skills"].append(entry)
478 self._save_installed(data)
480 def _load_installed(self) -> dict:
481 MARKET_DIR.mkdir(parents=True, exist_ok=True)
482 if not INSTALLED_JSON.exists():
483 return {"version": "1.0", "skills": []}
484 try:
485 return json.loads(INSTALLED_JSON.read_text(encoding="utf-8"))
486 except (json.JSONDecodeError, Exception):
487 return {"version": "1.0", "skills": []}
489 def _save_installed(self, data: dict):
490 MARKET_DIR.mkdir(parents=True, exist_ok=True)
491 INSTALLED_JSON.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
493 def _ensure_dirs(self):
494 MARKET_DIR.mkdir(parents=True, exist_ok=True)
495 SKILLS_DIR.mkdir(parents=True, exist_ok=True)