The Google Antigravity SDK is a Python SDK for building AI agents powered by Antigravity and Gemini. It provides a secure, scalable, and stateful infrastructure layer that abstracts the agentic loop, letting you focus on what your agent does rather than how it runs.
pip install google-antigravityImportant
The Google Antigravity SDK relies on a compiled runtime binary that is
included in the platform-specific wheels published to
PyPI. Cloning this
repository alone is not sufficient to run the SDK. Always install from
PyPI with pip install google-antigravity to obtain the binary.
Get started by running one of the examples/, such as the
hello_world example with:
export GEMINI_API_KEY="your_api_key_here"
python ./examples/getting_started/hello_world.pyThe Agent class is the easiest way to get started. It manages the full
lifecycle — binary discovery, tool wiring, hook registration, and policy
defaults — behind a single async context manager.
The system_instructions parameter is optional.
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())To stream agent output in real-time (e.g., for fluid UI or console applications), simply iterate over the ChatResponse object using an async for loop. The stream wrapper natively yields conversational str text tokens as they arrive, with zero network overhead:
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())For more complex use cases, you can also stream internal model reasoning/thinking or intercept tool call dispatches in real-time using dedicated async stream properties:
# 1. Stream reasoning/thinking deltas
async for thought in response.thoughts:
show_thinking_bubble(thought)
# 2. Stream strongly-typed ToolCall events
async for call in response.tool_calls:
show_executing_spinner(call.name)By default, Agent runs in read-only mode for safety. Pass
capabilities=CapabilitiesConfig() to enable all tools (including writes).
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)For full control over the connection lifecycle, use Conversation with a
ConnectionStrategy directly. Conversation is a stateful session that
accumulates step history, provides a chat() convenience method, and exposes
state introspection:
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:
# 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())Pass rich multimedia file attachments (images, videos, audio, and documents) to the agent alongside textual instruction prompt lists.
You can attach assets directly using content classes (perfect for in-memory bytes) or conveniently from a filesystem path (which automatically resolves types and guesses MIME formats):
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())Register Python functions as tools that the agent can call:
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?")Connect to external MCP servers and expose their tools to the agent:
from google.antigravity import Agent, LocalAgentConfig
from google.antigravity.types import McpStdioServer
config = LocalAgentConfig(
mcp_servers=[McpStdioServer(command="npx", args=["my-mcp-server"])],
)
async with Agent(config) as agent:
response = await agent.chat("Use the MCP tools to help me.")Control agent behavior with a declarative policy system:
from google.antigravity import Agent, 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,
)
async with Agent(config) as agent:
await run_interactive_loop(agent)Run background tasks that react to external events and push messages into the agent:
from google.antigravity import Agent, 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)],
)
async with Agent(config) as agent:
await run_interactive_loop(agent)The SDK follows a three-layer architecture:
| Layer | Purpose | Key Classes |
|---|---|---|
| Layer 1 — Simplified | High-level, batteries-included entry point | Agent |
| Layer 2 — Session | Stateful session with history and convenience methods | Conversation, ChatResponse, Step, ToolCall, AgentConfig, HookRunner, ToolRunner, TriggerRunner |
| Layer 3 — Adapter | Transport and backend abstraction | Connection, ConnectionStrategy, LocalConnection |
For more detailed documentation on specific components, see:
- Agent — High-level, batteries-included entry point.
- Connections — Transport and backend abstraction.
- Conversation — Stateful session management.
- Hooks — Agent lifecycle interception and policies.
- MCP — Model Context Protocol integration.
- Tools — In-process tool execution.
- Triggers — Background tasks and external events.