Coverage for agentos/marketplace/registry.py: 0%
262 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-02 09:59 +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 stats(self) -> dict:
228 """市场统计。"""
229 installed = self.list_installed()
230 by_format = {}
231 for m in installed:
232 fmt = m.format.value
233 by_format[fmt] = by_format.get(fmt, 0) + 1
234 return {
235 "total": len(installed),
236 "by_format": by_format,
237 "market_dir": str(MARKET_DIR),
238 }
240 def _check_duplicate(self, name: str) -> Optional[InstallResult]:
241 """如果 skill 已安装，返回失败结果；否则返回 None。"""
242 existing = self.get_installed(name)
243 if existing:
244 return InstallResult(
245 success=False,
246 error=f"Skill '{name}' already installed (v{existing.version}). Use 'marketplace update {name}' to upgrade.",
247 )
248 return None
250 def _install_pypi(self, name: str) -> InstallResult:
251 """从 PyPI 安装 agentos-skill-<name>。"""
252 pkg = f"{PYPI_SKILL_PREFIX}{name}"
253 try:
254 r = subprocess.run(
255 [sys.executable, "-m", "pip", "install", pkg, "--quiet", "--disable-pip-version-check"],
256 capture_output=True, text=True, timeout=120,
257 )
258 if r.returncode != 0:
259 return InstallResult(success=False, error=f"pip install failed: {r.stderr[:200]}")
260 except subprocess.TimeoutExpired:
261 return InstallResult(success=False, error="pip install timed out")
263 # 查找安装后的包位置，读取 manifest
264 manifest = self._find_package_manifest(pkg)
265 if not manifest:
266 # 没有 manifest 的 PyPI 包也注册为 generic skill
267 manifest = SkillManifest(
268 name=name,
269 version="?",
270 description=f"PyPI package: {pkg}",
271 format=SkillFormat.GENERIC,
272 source="pypi",
273 )
274 manifest.source = "pypi"
276 # 复制到 skills 目录
277 self._copy_to_skills(name, manifest)
279 # 注册
280 self._register_manifest(manifest)
281 return InstallResult(
282 success=True,
283 manifest=manifest,
284 install_type="pypi_install",
285 pypi_package=pkg,
286 )
288 def _install_local(self, local_path: Path, manifest_file: Path) -> InstallResult:
289 """从本地目录安装。"""
290 raw = manifest_file.read_text(encoding="utf-8")
291 if manifest_file.suffix in (".yaml", ".yml"):
292 import yaml
293 data = yaml.safe_load(raw) or {}
294 else:
295 data = json.loads(raw)
297 manifest = SkillManifest.from_dict(data, source="local", install_path=str(local_path))
298 name = manifest.name or local_path.name
300 dup = self._check_duplicate(name)
301 if dup:
302 return dup
304 # 安装依赖
305 deps = self._install_deps(manifest)
307 # 复制到 skills 目录
308 dest = SKILLS_DIR / name
309 if dest.exists():
310 shutil.rmtree(dest)
311 shutil.copytree(local_path, dest)
313 manifest.install_path = str(dest)
314 manifest.source = "local"
315 self._register_manifest(manifest)
317 return InstallResult(
318 success=True,
319 manifest=manifest,
320 install_type="local_copy",
321 dep_installed=deps,
322 )
324 def _install_github(self, url: str) -> InstallResult:
325 """从 GitHub 克隆安装。"""
326 name = url.rstrip("/").split("/")[-1].replace(".git", "")
327 dest = SKILLS_DIR / name
329 if dest.exists():
330 shutil.rmtree(dest)
332 try:
333 r = subprocess.run(
334 ["git", "clone", "--depth=1", url, str(dest)],
335 capture_output=True, text=True, timeout=60,
336 )
337 if r.returncode != 0:
338 return InstallResult(success=False, error=f"git clone failed: {r.stderr[:200]}")
339 except FileNotFoundError:
340 return InstallResult(success=False, error="git not found. Install git first.")
341 except subprocess.TimeoutExpired:
342 return InstallResult(success=False, error="git clone timed out")
344 # 查找 manifest 文件
345 manifest_file = None
346 for fname in ("skill.yaml", "skill.json", "manifest.yaml", "manifest.json"):
347 candidate = dest / fname
348 if candidate.exists():
349 manifest_file = candidate
350 break
352 if manifest_file:
353 raw = manifest_file.read_text(encoding="utf-8")
354 if manifest_file.suffix in (".yaml", ".yml"):
355 import yaml
356 data = yaml.safe_load(raw)
357 else:
358 data = json.loads(raw)
359 manifest = SkillManifest.from_dict(data, source="github", install_path=str(dest))
360 else:
361 manifest = SkillManifest(
362 name=name,
363 version="0.1.0",
364 description=f"GitHub skill: {url}",
365 format=SkillFormat.GENERIC,
366 source="github",
367 install_path=str(dest),
368 repository=url,
369 )
371 manifest.source = "github"
372 manifest.repository = url
373 if not manifest.name:
374 manifest.name = name
376 deps = self._install_deps(manifest)
377 self._register_manifest(manifest)
379 return InstallResult(
380 success=True,
381 manifest=manifest,
382 install_type="git_clone",
383 dep_installed=deps,
384 )
386 def _install_deps(self, manifest: SkillManifest) -> list[str]:
387 """安装 skill 的 pip 依赖，返回成功安装的包名列表。"""
388 installed = []
389 for dep in manifest.dependencies:
390 try:
391 subprocess.run(
392 [sys.executable, "-m", "pip", "install", dep, "--quiet", "--disable-pip-version-check"],
393 capture_output=True, timeout=60,
394 )
395 installed.append(dep)
396 except Exception:
397 pass
398 return installed
400 def _find_package_manifest(self, pkg_name: str) -> Optional[SkillManifest]:
401 """从已安装的 PyPI 包中查找 manifest。"""
402 try:
403 r = subprocess.run(
404 [sys.executable, "-m", "pip", "show", "-f", pkg_name],
405 capture_output=True, text=True, timeout=10,
406 )
407 if r.returncode != 0:
408 return None
410 # 找 Location 行
411 location = ""
412 for line in r.stdout.split("\n"):
413 if line.startswith("Location:"):
414 location = line.split(":", 1)[1].strip()
415 break
417 if not location:
418 return None
420 # 尝试常见 manifest 路径
421 pkg_name_clean = pkg_name.replace("-", "_")
422 candidates = [
423 Path(location) / pkg_name_clean / "skill.yaml",
424 Path(location) / pkg_name_clean / "skill.json",
425 Path(location) / pkg_name_clean / "manifest.yaml",
426 Path(location) / pkg_name_clean / "manifest.json",
427 ]
428 for p in candidates:
429 if p.exists():
430 raw = p.read_text(encoding="utf-8")
431 if p.suffix in (".yaml", ".yml"):
432 import yaml
433 data = yaml.safe_load(raw)
434 else:
435 data = json.loads(raw)
436 return SkillManifest.from_dict(data, source="pypi")
437 except Exception:
438 pass
439 return None
441 def _copy_to_skills(self, name: str, manifest: SkillManifest):
442 """确保 skill 源文件在 skills 目录有一份。"""
443 dest = SKILLS_DIR / name
444 dest.mkdir(parents=True, exist_ok=True)
445 manifest_path = dest / "manifest.yaml"
446 import yaml
447 manifest_path.write_text(
448 yaml.dump(manifest.to_dict(), allow_unicode=True, default_flow_style=False, sort_keys=False),
449 encoding="utf-8",
450 )
452 def _register_manifest(self, manifest: SkillManifest):
453 """将 manifest 注册到 installed.json。"""
454 data = self._load_installed()
455 # 去重
456 data["skills"] = [s for s in data.get("skills", []) if s.get("name") != manifest.name]
457 entry = manifest.to_dict()
458 entry["installed_at"] = time.time()
459 data["skills"].append(entry)
460 self._save_installed(data)
462 def _load_installed(self) -> dict:
463 MARKET_DIR.mkdir(parents=True, exist_ok=True)
464 if not INSTALLED_JSON.exists():
465 return {"version": "1.0", "skills": []}
466 try:
467 return json.loads(INSTALLED_JSON.read_text(encoding="utf-8"))
468 except (json.JSONDecodeError, Exception):
469 return {"version": "1.0", "skills": []}
471 def _save_installed(self, data: dict):
472 MARKET_DIR.mkdir(parents=True, exist_ok=True)
473 INSTALLED_JSON.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
475 def _ensure_dirs(self):
476 MARKET_DIR.mkdir(parents=True, exist_ok=True)
477 SKILLS_DIR.mkdir(parents=True, exist_ok=True)