antigravity-acp-sdk 0.1.2

A generic client and connection pool implementation for Agent Client Protocol (ACP)

Antigravity ACP SDK

antigravity-acp-sdk 是基于官方 agent-client-protocol (ACP) 实现的通用 Python 客户端与长连接并发池管理器。

安装方式

1. 本地联调模式 (开发时推荐)

在主项目的 pyproject.toml 中配置本地相对路径映射：

[project]
dependencies = [
    "antigravity-acp-sdk",
]

[tool.uv.sources]
antigravity-acp-sdk = { path = "antigravity-acp-sdk", editable = true }

2. Git 依赖模式

直接指向远程 Git 仓库：

[project]
dependencies = [
    "antigravity-acp-sdk @ git+ssh://[e-mail]/yourorg/antigravity-acp-sdk.git@v0.1.0"
]

系统环境依赖与安装

本包运行依赖于 agy CLI 命令行工具以及 agy-acp 通信桥接可执行程序。任何引入本包的项目环境均需要预先下载并安装这两个系统级依赖。

1. 安装官方 agy CLI

agy 是运行智能体的底层命令行工具。可以通过官方安装脚本进行安装：

curl -fsSL https://antigravity.google/cli/install.sh | bash
# 默认会安装到 ~/.local/bin/agy。为了让系统全局可用，建议将其移动到 /usr/local/bin：
# mv ~/.local/bin/agy /usr/local/bin/agy

2. 下载并安装 agy-acp

agy-acp 是连接池直接拉起长连接的 stdio 桥接执行文件：

• 下载地址：https://www.fentaiq.com:543/agy-acp
• 安装命令：

  curl -fsSL https://www.fentaiq.com:543/agy-acp -o /usr/local/bin/agy-acp
  chmod +x /usr/local/bin/agy-acp
  

3. 环境与 PATH 变量配置

默认情况下，连接池会在系统的 PATH 环境变量中查找 agy-acp。

• 如果 agy-acp 被安装到标准的 /usr/local/bin 中，连接池可自动直接调用。
• 如果安装在非标准路径（例如本地 Python 虚拟环境 .venv/bin 或用户的本地二进制目录 ~/.local/bin），则必须在初始化连接池时通过 extra_paths 参数将该目录传入，例如 extra_paths=["/home/user/.local/bin"]，否则会抛出子进程启动找不到命令的错误。

环境变量与运行配置

本包在拉起 ACP 子进程时，会自动同步并支持以下关键环境变量以控制智能体的底层行为。使用本包的项目应在启动前配置这些环境变量：

1. 模型指定 (AGY_MODEL)：
   • 通过环境变量 AGY_MODEL 设定 ACP 进程的默认分析模型（如 "Gemini 3.5 Flash (Medium)"）。
2. ACP 额外运行参数 (AGY_EXTRA_ARGS)：
   • 默认会自动注入参数：--dangerously-skip-permissions --print-timeout=10m0s。这对于自动化测试和无人工介入运行（Non-interactive）场景非常关键，用以绕过命令行交互权限确认并防止指令超时。

智能体配置与资产结构规范 (By Convention)

本 SDK 采用**约定优于配置（Convention over Configuration）**的策略：

1. 目录即智能体命名约定（强制要求）

所有智能体必须在统一的 agent/ 文件夹下进行组织，子目录名称必须作为智能体的名称，且必须包含 AGENTS.md 作为合法智能体标识：

host_project/
└── agent/
    ├── mcp_config.json               # 1. 全局默认 MCP 配置文件 (所有 Agent 共享)
    ├── hooks.json                    # 2. 全局默认 Hooks 配置文件 (所有 Agent 共享)
    ├── access_control_rules.json     # 3. 全局默认访问控制阻断文件 (所有 Agent 共享)
    ├── scripts/                      # 4. 全局公用钩子脚本目录
    │   ├── _shared.py
    │   ├── validate.py
    │   ├── validate_delivery.py
    │   └── check_file_access.py
    └── stock_analyzer/               # 5. 智能体专属配置目录 (名称: stock_analyzer)
        ├── AGENTS.md                 # 专属规则 (Rules - 必须存在)
        ├── mcp_config.json           # 专属 MCP 配置文件 (可选)
        ├── hooks.json                # 专属规则挂钩文件 (可选)
        ├── access_control_rules.json # 专属访问控制挂钩文件 (可选)
        └── skills/                   # 专属技能目录 (Skills)
            └── stock_analysis/

2. 三层配置深度合并机制 (Triple-Layered Deep Merge)

在启动工作区时，SDK 会自动将三层配置进行深度合并，生成最终沙箱 <cwd>/.agents/ 下对应的配置文件：

• JSON 配置合并语义 (Overwrite)：
  • 字典（dict）类型：执行递归深度合并。相同 key 值以高优先级层级覆盖。
  • 非字典（如 list、scalar）类型：直接覆盖（Overwrite），高优先级配置整段替换低优先级。
• 同名文件夹合并：
  • 三层中同名文件夹（如 skills/、scripts/ 等资源目录）会递归拷贝并合并至沙箱。
  • 如果拷贝中发现同名文件冲突，拷贝过程将产生警告记录。

---

核心接口说明

1. 连接池初始化 (极简无参化)

from antigravity_acp_sdk import AcpConnectionPool

pool = AcpConnectionPool.get_instance(
    max_connections=4,
    command="agy-acp",
    extra_paths=["/path/to/custom/bin"]
)

2. 获取连接与输入编译 (client.new_session)

支持传入 inputs 字典对规则和技能的 Markdown 模板进行安全变量编译（支持可选变量 {{ x | default('') }}），并自动处理文件/目录输入的拷贝与关联。

输入处理策略：

1. 单个文件：自动拷贝至沙箱 .agents/input/ 目录下，并使用其相对路径替换模板变量。
2. 目录路径：
   • 默认行为（整体递归拷贝）：将整个目录及其内容递归复制到 .agents/input/<dirname> 目录下，保障智能体无法篡改宿主机源数据，具备强写隔离安全性。
   • 可选行为（软链接模式）：当指定参数 link_inputs=True 时，将在沙箱内创建指向源目录的软链接（若系统不支持或软链接失败，将自动回退到整体拷贝模式），适用于超大目录且无写隔离要求的场景。

from antigravity_acp_sdk import AgentManager

# 初始化智能体定义
manager = AgentManager("agent")
agent = manager.get_agent("stock_analyzer")

# 申请连接
client = await pool.acquire()

# 开启会话，传入 inputs 字典自动处理拷贝与模板渲染
session_resp = await client.new_session(
    cwd="/path/to/sandbox",
    agent=agent,
    inputs={
        "financial_data": "/abs/path/to/annual_financial_metrics.tsv", # 文件：将自动拷贝到 .agents/input/ 并替换为相对路径
        "raw_dataset_dir": "/abs/path/to/raw_dataset",                 # 目录：默认会被整体递归拷贝到 .agents/input/ 下
        "analysis_guideline": "注重对资产流动性指标的审核"             # 文本：直接执行模板变量替换
    },
    link_inputs=False  # 可选：设为 True 以启用软链接模式
)
session_id = session_resp.session_id

3. 一键运行与交付物自动回收 (client.prompt)

宿主使用 client.prompt 启动智能体。SDK 会在内存中读取并自动解析 .agents/output/ 目录下的所有产出（封装为 AgentOutputs 对象），随即物理清空 input/ 与 output/ 临时资源目录，宿主无须关心清理细节。

from acp import text_block

# 运行并回收成果
prompt_resp, outputs = await client.prompt(
    session_id=session_id,
    prompt=[text_block("Use skill: stock_analysis, 分析股票 000001.SZ")]
)

# 1. 自动解析为 dict (针对 .json)
report = outputs.get_content("report.json")
# 2. 读取为普通文本 (针对其他格式)
text_log = outputs.get_content("sub_folder/logs.txt")

---

命令行开发辅助与诊断工具 (Bin CLI Tools)

1. verify-mcp：MCP 连通性一键诊断

一键对三层合并后的特定智能体 MCP 进程执行 standard 初始化握手诊断：

uv run verify-mcp stock_analyzer --base-dir agent/ --extra-paths .venv/bin

2. inspect-agent：智能体配置及沙箱结构大盘点

交互式盘点其规则、技能和 Hooks，输出生成的 .agents/ 目标沙箱结构：

uv run inspect-agent --base-dir agent/ --extra-paths .venv/bin

3. sync-schema：Pydantic 类 Schema 导出

将开发环境定义的 Pydantic 类一键同步至智能体 schemas 目录下：

uv run sync-schema -f src/stock_data_fetcher/analysis/models.py -c StockAnalysisReport -o agent/stock_analyzer/schemas/report.schema.json