google-antigravity/antigravity-sdk-python
GitHub: google-antigravity/antigravity-sdk-python
一个基于 Python 的 AI agent 构建框架，由 Gemini 和 Google Antigravity 驱动，提供有状态会话管理、工具调用和多模态处理能力。
Stars: 2431 | Forks: 910
# Google Antigravity SDK
Google Antigravity SDK 是一个 Python SDK，用于构建由 Antigravity 和 Gemini 驱动的 AI agent。它提供了一个安全、可扩展且有状态的基础设施层，抽象了 agentic loop，让您可以专注于 agent *做什么*，而不是它如何运行。
## 安装
```
pip install google-antigravity
```
## 快速开始
通过运行 [`examples/`](examples/) 中的示例来入门，例如使用以下命令运行 `hello_world` 示例：
```
export GEMINI_API_KEY="your_api_key_here"
python ./examples/getting_started/hello_world.py
```
## Gemini Enterprise Agent Platform（原 Vertex AI）
要将 SDK 与 Gemini Enterprise Agent Platform（原 Vertex AI）配合使用，请使用 `vertex=True` 配置 `LocalAgentConfig`，并指定您的 GCP `project` 和 `location`。
默认情况下，SDK 使用 Application Default Credentials (ADC) 进行身份验证。
```
from google.antigravity import Agent, LocalAgentConfig
config = LocalAgentConfig(
vertex=True,
project="your-gcp-project",
location="us-central1",
)
async with Agent(config) as agent:
response = await agent.chat("Hello!")
print(await response.text())
```
在运行 agent 之前，请确保您已在本地完成身份验证：
```
gcloud auth application-default login
```
## 概念
### 简单 Agent
`Agent` 类是最简单的入门方式。它在一个异步 context manager 背后管理着完整的生命周期——二进制发现、工具连接、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` 对象。stream wrapper 会在对话 `str` 文本 token 到达时原生地将其输出，且没有任何网络开销：
```
import asyncio
import sys
from google.antigravity import Agent, LocalAgentConfig
async def main():
config = LocalAgentConfig()
async with Agent(config) as agent:
# Returns instantly — does not block
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())
```
### Sugar 化 Thoughts 和 Tool Call 流（高级）
对于更复杂的用例，您还可以使用专用的 async stream 属性，流式传输内部模型推理/thinking 或实时拦截 tool call 分发：
```
# 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 LocalAgentConfig, CapabilitiesConfig
from google.antigravity.utils.interactive import run_interactive_loop
config = LocalAgentConfig(
# api_key="your_api_key_here",
capabilities=CapabilitiesConfig(),
)
await run_interactive_loop(config)
```
### Conversation 的高级用法
要完全控制连接生命周期，请直接使用带有 `ConnectionStrategy` 的 `Conversation`。`Conversation` 是一个有状态的 session，它会累积 step 历史记录，提供便捷的 `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
async def main():
tool_runner = ToolRunner()
strategy = LocalConnectionStrategy(
tool_runner=tool_runner,
)
async with Conversation.create(strategy) as conversation:
# High-level: one-call send + collect
response = await conversation.chat("What files are here?")
print(await response.text())
# Step history accumulates automatically
print(f"Total steps: {len(conversation.history)}")
print(f"Turns: {conversation.turn_count}")
print(f"Last response: {conversation.last_response}")
# Low-level: streaming steps
await conversation.send("Tell me more.")
async for step in conversation.receive_steps():
if step.is_complete_response:
print(step.content)
asyncio.run(main())
```
## 功能
### 多模态摄取
将丰富的多媒体文件附件（图像、视频、音频和文档）与文本 instruction prompt 列表一起传递给 agent。
您可以**直接使用内容类附加资产**（非常适合内存中的 bytes），也可以**方便地从文件系统路径附加**（它会自动解析类型并猜测 MIME 格式）：
```
from google.antigravity import Agent, LocalAgentConfig
from google.antigravity.types import Image, from_file
config = LocalAgentConfig(system_instructions="You are an expert software architect.")
async with Agent(config) as agent:
# 1. Flat filesystem shortcut (automatically resolves as types.Document)
pdf_spec = from_file("spec.pdf")
# 2. Direct constructor instantiation (perfect for in-memory raw bytes)
chart_image = Image(
data=b"raw_png_bytes_here",
mime_type="image/png",
description="Architecture blueprint"
)
# Send a mixed list of text instructions and content classes
prompt = [
"Analyze this chart against the specification and list three security vulnerabilities:",
chart_image,
pdf_spec
]
response = await agent.chat(prompt)
print(await response.text())
```
### 自定义工具
将 Python 函数注册为 agent 可以调用的工具：
```
def get_weather(city: str) -> str:
"""Returns the current weather for a city."""
return f"It's sunny in {city}."
config = LocalAgentConfig(
tools=[get_weather],
)
async with Agent(config) as agent:
response = await agent.chat("What's the weather in Tokyo?")
```
### MCP 集成
连接到外部 [MCP](https://modelcontextprotocol.io/) 服务器，并将其工具暴露给 agent：
```
from google.antigravity import Agent, LocalAgentConfig
from google.antigravity.types import McpStdioServer
config = LocalAgentConfig(
mcp_servers=[McpStdioServer(name="my_server", command="npx", args=["my-mcp-server"])],
)
async with Agent(config) as agent:
response = await agent.chat("Use the MCP tools to help me.")
```
### Hooks 和策略
使用声明式策略系统控制 agent 行为：
```
from google.antigravity import LocalAgentConfig, CapabilitiesConfig
from google.antigravity.hooks.policy import deny, allow, ask_user, enforce
from google.antigravity.utils.interactive import run_interactive_loop
policies = [
deny("*"), # Block all tools by default
allow("view_file"), # Allow reading files
ask_user("run_command", handler=my_handler), # Ask before running commands
]
config = LocalAgentConfig(
capabilities=CapabilitiesConfig(),
policies=policies,
)
await run_interactive_loop(config)
```
### 触发器
运行响应外部事件的后台任务，并将消息推送给 agent：
```
from google.antigravity import LocalAgentConfig
from google.antigravity.triggers import every
from google.antigravity.utils.interactive import run_interactive_loop
async def check_status(ctx):
await ctx.send("Check the deployment status.")
config = LocalAgentConfig(
triggers=[every(60, check_status)],
)
await run_interactive_loop(config)
```
## 架构
SDK 遵循三层架构：
| 层 | 用途 | 关键类 |
|:------|:--------|:------------|
| **第 1 层** — 简化 | 高级、开箱即用的入口点 | `Agent` |
| **第 2 层** — Session | 具有 history 和便捷方法的有状态 session | `Conversation`, `ChatResponse`, `Step`, `ToolCall`, `AgentConfig`, `HookRunner`, `ToolRunner`, `TriggerRunner` |
| **第 3 层** — Adapter | 传输和后端抽象 | `Connection`, `ConnectionStrategy`, `LocalConnection` |
## 组件文档
有关特定组件的更多详细文档，请参阅：
- [Agent](google/antigravity/agent.py) — 高级、开箱即用的入口点。
- [Connections](google/antigravity/connections/README.md) — 传输和后端抽象。
- [Conversation](google/antigravity/conversation/README.md) — 有状态的 session 管理。
- [Hooks](google/antigravity/hooks/README.md) — Agent 生命周期拦截和策略。
- [MCP](google/antigravity/mcp/README.md) — Model Context Protocol 集成。
- [Tools](google/antigravity/tools/README.md) — 进程内工具执行。
- [Triggers](google/antigravity/triggers/README.md) — 后台任务和外部事件。
## 许可证
[Apache License 2.0](LICENSE)
标签：AI智能体, GCP, Gemini, Python, 大模型SDK, 开发框架, 无后门, 计算机取证, 逆向工具