Google Antigravity SDK

Google Antigravity SDK 是一个 Python SDK，用于构建由 Antigravity 和 Gemini 驱动的 AI Agent。它提供了一个安全、可扩展且具备状态的基础设施层，能够抽象出 Agent 的执行循环，让您可以专注于 Agent 的“业务逻辑”，而无需关心其底层运行机制。

安装

pip install google-antigravity

[!IMPORTANT] Google Antigravity SDK 依赖于编译好的运行时二进制文件，该文件已包含在发布到 PyPI 的特定于平台的 wheel 包中。仅克隆此仓库不足以运行 SDK。 请务必始终通过 pip install google-antigravity 从 PyPI 安装以获取该二进制文件。

快速入门

运行 examples/ 中的示例程序即可开始，例如运行 hello_world 示例：

export GEMINI_API_KEY="your_api_key_here"
python ./examples/getting_started/hello_world.py

核心概念

简单 Agent

Agent 类是上手最简单的方式。它通过单个异步上下文管理器管理完整的生命周期——包括二进制文件发现、工具连接、钩子（hook）注册以及策略默认值配置。

system_instructions 参数为可选。

import asyncio
from google.antigravity import Agent, LocalAgentConfig

async def main():
    config = LocalAgentConfig(
        system_instructions="You are an expert assistant for codebase navigation.",
        # api_key="your_api_key_here",
    )
    async with Agent(config) as agent:
        response = await agent.chat("What files are in the current directory?")
        print(await response.text())

async def run():
    await main()

if __name__ == "__main__":
    asyncio.run(run())

流式响应

若要实时流式传输 Agent 的输出（例如用于流畅的 UI 或控制台应用程序），只需使用 async for 循环遍历 ChatResponse 对象即可。流包装器会在令牌到达时原生生成对话 str 文本令牌，且几乎没有网络开销：

import asyncio
import sys
from google.antigravity import Agent, LocalAgentConfig

async def main():
    config = LocalAgentConfig()
    async with Agent(config) as agent:
        # 立即返回 — 不会阻塞
        response = await agent.chat("Write a short poem about space.")
        
        async for token in response:
            sys.stdout.write(token)
            sys.stdout.flush()
        print()

asyncio.run(main())

思考过程与工具调用流（高级）

对于更复杂的用例，您还可以使用专用的异步流属性实时流式传输模型的内部推理/思考过程，或拦截工具调用的分发：

# 1. 流式传输推理/思考增量
async for thought in response.thoughts:
    show_thinking_bubble(thought)

# 2. 流式传输强类型的 ToolCall 事件
async for call in response.tool_calls:
    show_executing_spinner(call.name)

出于安全考虑，Agent 默认以只读模式运行。如需启用所有工具（包括写操作），请传入 capabilities=CapabilitiesConfig()。

交互循环

from google.antigravity import Agent, LocalAgentConfig, CapabilitiesConfig
from google.antigravity.utils.interactive import run_interactive_loop

config = LocalAgentConfig(
    # api_key="your_api_key_here",
    capabilities=CapabilitiesConfig(),
)
async with Agent(config) as agent:
    await run_interactive_loop(agent)

高级会话用法

若要完全控制连接生命周期，请直接配合 ConnectionStrategy 使用 Conversation。Conversation 是一个有状态的会话，它会累积步骤历史记录，提供便捷的 chat() 方法，并公开状态内省功能：

import asyncio
from google.antigravity.connections.local import LocalConnectionStrategy
from google.antigravity.conversation.conversation import Conversation
from google.antigravity.tools.tool_runner import ToolRunner
from google.antigravity.types import GeminiConfig

async def main():
    tool_runner = ToolRunner()
    strategy = LocalConnectionStrategy(
        tool_runner=tool_runner,
        # gemini_config=GeminiConfig(api_key="your_api_key_here"),
    )
    
    async with Conversation.create(strategy) as conversation:
        # 高级用法：一次性发送 + 收集
        response = await conversation.chat("What files are here?")
        print(await response.text())
        
        # 步骤历史记录会自动累积
        print(f"Total steps: {len(conversation.history)}")
        print(f"Turns: {conversation.turn_count}")
        print(f"Last response: {conversation.last_response}")
        
        # 低级用法：流式处理步骤
        await conversation.send("Tell me more.")
        async for step in conversation.receive_steps():
            if step.is_complete_response:
                print(step.content)

asyncio.run(main())

功能特性

多模态输入

传入丰富的多媒体文件附件（图像、视频、音频和文档）。