lambdagent/scripts/install_builtin_packs.py

#!/usr/bin/env python3
"""
安装仓库内置的 AgentPack 到本地 agentpacks 目录。

用法（server 必须先运行）：
    python scripts/install_builtin_packs.py

也可以不启动 server，直接写入 agentpacks_dir：
    python scripts/install_builtin_packs.py --offline

内置 pack 位置：agentexample/agentpacks/
安装目标：    ~/.agentpaas/data/agentpacks/   (desktop 模式)
              ~/LambdAgentDesktop/agentpacks/  (如果是 desktop 模式且未改数据目录)
"""
from __future__ import annotations

import argparse
import json
import os
import sys
import zipfile
import tempfile
from pathlib import Path


REPO_ROOT = Path(__file__).parent.parent.resolve()
PACKS_SRC  = REPO_ROOT / "agentexample" / "agentpacks"

BUILTIN_PACKS = [
    "research.top-journal-reviewer",
    "research.literature-mapper",
    "research.grant-planner",
]


def _zip_pack(pack_id: str, dest_dir: Path) -> str:
    pack_dir = PACKS_SRC / pack_id
    if not pack_dir.is_dir():
        sys.exit(f"❌  Pack 目录不存在: {pack_dir}")
    zip_path = str(dest_dir / f"{pack_id}.zip")
    with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
        for p in pack_dir.rglob("*"):
            if p.is_file():
                zf.write(p, p.relative_to(pack_dir))
    return zip_path


def _default_packs_dir() -> str:
    """读取 ~/.agentpaas/config.json 或按 desktop 模式默认值推断 agentpacks_dir。"""
    cfg = Path.home() / ".agentpaas" / "config.json"
    try:
        with open(cfg) as f:
            data = json.load(f) or {}
        data_dir = data.get("data_dir", "")
    except Exception:
        data_dir = ""
    if not data_dir:
        # desktop 默认
        data_dir = str(Path.home() / "LambdAgentDesktop")
    return os.path.join(data_dir, "agentpacks")


# ── 离线模式：直接调用 install_from_zip ──────────────────────────────────────

def install_offline(packs_dir: str, packs: list[str]) -> None:
    from agentpaas.engine.agentpack_store import install_from_zip, get_installed

    os.makedirs(packs_dir, exist_ok=True)
    print(f"\n安装目标目录: {packs_dir}\n")

    with tempfile.TemporaryDirectory() as td:
        for pack_id in packs:
            existing = get_installed(packs_dir, pack_id)
            if existing:
                print(f"  ✓  {pack_id} v{existing.version}  — 已安装，跳过")
                continue
            zip_path = _zip_pack(pack_id, Path(td))
            try:
                pack = install_from_zip(zip_path, packs_dir)
                print(f"  ✓  {pack_id} v{pack.version}  — 安装完成")
            except Exception as e:
                print(f"  ✗  {pack_id}  — 失败: {e}", file=sys.stderr)


# ── 在线模式：通过 HTTP API 安装（需要 server 运行）───────────────────────────

def install_online(base_url: str, api_key: str, packs: list[str]) -> None:
    try:
        import requests
    except ImportError:
        sys.exit("缺少依赖: pip install requests")

    session = requests.Session()
    if api_key:
        session.headers["Authorization"] = f"Bearer {api_key}"

    print(f"\n连接 server: {base_url}\n")

    # 先列出已安装
    try:
        r = session.get(f"{base_url}/api/v1/agentpacks", timeout=5)
        r.raise_for_status()
        already = {p["id"] for p in r.json().get("agentpacks", [])}
    except Exception as e:
        sys.exit(f"❌  无法连接 server: {e}\n    请先启动: uvicorn agentpaas.api.app:app --port 8000")

    with tempfile.TemporaryDirectory() as td:
        for pack_id in packs:
            if pack_id in already:
                print(f"  ✓  {pack_id}  — 已安装，跳过")
                continue
            zip_path = _zip_pack(pack_id, Path(td))
            r = session.post(
                f"{base_url}/api/v1/agentpacks/install",
                json={"zip_path": zip_path},
                timeout=30,
            )
            if r.status_code == 200:
                pack = r.json()["installed"]
                print(f"  ✓  {pack['id']} v{pack['version']}  — 安装完成")
            else:
                print(f"  ✗  {pack_id}  — HTTP {r.status_code}: {r.text[:120]}", file=sys.stderr)


# ── CLI ───────────────────────────────────────────────────────────────────────

def main() -> None:
    parser = argparse.ArgumentParser(
        description="安装仓库内置 AgentPack (research.* 三个 pack)",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例:
  # 离线安装（不需要 server 运行，推荐）
  python scripts/install_builtin_packs.py --offline

  # 在线安装（需要 server 先跑起来）
  python scripts/install_builtin_packs.py

  # 指定安装目录
  python scripts/install_builtin_packs.py --offline --packs-dir /data/agentpacks
""",
    )
    parser.add_argument(
        "--offline", action="store_true",
        help="直接写入 agentpacks_dir，不需要 server 运行（推荐）",
    )
    parser.add_argument(
        "--packs-dir", default="",
        help="安装目标目录（默认：从 ~/.agentpaas/config.json 推断）",
    )
    parser.add_argument(
        "--url", default="http://127.0.0.1:8000",
        help="Server URL（在线模式，默认 http://127.0.0.1:8000）",
    )
    parser.add_argument(
        "--key", default="",
        help="API key（在线模式，默认从 ~/.agentpaas/config.json 读取）",
    )
    parser.add_argument(
        "--pack", action="append", dest="packs",
        help="只安装指定 pack（可多次指定，默认全部三个）",
    )
    args = parser.parse_args()

    packs = args.packs or BUILTIN_PACKS
    print(f"内置 Pack 源目录: {PACKS_SRC}")

    if args.offline:
        packs_dir = args.packs_dir or _default_packs_dir()
        install_offline(packs_dir, packs)
    else:
        # 读取 api_key
        api_key = args.key
        if not api_key:
            cfg = Path.home() / ".agentpaas" / "config.json"
            try:
                api_key = (json.load(open(cfg)) or {}).get("api_key", "")
            except Exception:
                pass
        install_online(args.url, api_key, packs)

    print('\n完成。在 webui 的"智能体包"页面可以看到已安装的 pack，点击"创建智能体"即可使用。\n')


if __name__ == "__main__":
    main()