Google Antigravity Python SDK

Google's official Python SDK for building AI agents powered by antigravity-2.0 and Gemini. Published as google-antigravity on PyPI (v0.1.2). Requires Python >= 3.10. Licensed under Apache 2.0.

Repository: github.com/Google-Antigravity/antigravity-sdk-python

---

Package Structure

google/antigravity/
├── __init__.py              # Public API exports
├── agent.py                 # Layer 1: Agent (high-level entry point)
├── types.py                 # Canonical Pydantic V2 boundary types
├── agent_test.py            # Agent API tests
├── connections/
│   ├── connection.py        # ABCs: Connection, ConnectionStrategy, AgentConfig
│   ├── local/
│   │   ├── local_connection.py         # WebSocket connection to Go harness
│   │   ├── local_connection_config.py  # LocalAgentConfig
│   │   ├── localharness_pb2.py         # Protobuf bindings
│   │   └── types.py                    # Structured tool result types
│   └── README.md
├── conversation/
│   ├── conversation.py      # Layer 2: Conversation (session state)
│   └── README.md
├── tools/
│   ├── tool_runner.py       # Registry + executor for Python callables
│   ├── tool_context.py      # Conversation-aware context for tools
│   └── README.md
├── hooks/
│   ├── hooks.py             # Hook base classes and decorator factories
│   ├── hook_runner.py       # Dispatch engine
│   ├── policy.py            # Declarative policy system (allow/deny/ask_user)
│   └── README.md
├── triggers/
│   ├── triggers.py          # Trigger type + TriggerContext
│   ├── trigger_runner.py    # Lifecycle management
│   ├── helpers.py           # every(), on_file_change() factories
│   └── README.md
├── utils/
│   └── interactive.py       # CLI REPL, spinner, interactive hooks
└── bin/
    └── localharness          # Compiled Go binary (platform-specific wheel)

Dependencies

# pyproject.toml
dependencies = [
    "absl-py",
    "google-genai>=1.0",
    "mcp>=1.0",
    "pydantic>=2.0",
    "uvicorn>=0.46",
    "websockets>=12.0",
    "protobuf>=4.25",
]

The SDK ships a pre-compiled Go binary (localharness) bundled in platform-specific wheels. The binary is the actual agent runtime; the Python layer orchestrates it.

---

Three-Layer Architecture

graph TB
    subgraph L1["Layer 1 - Simplified"]
        Agent["Agent<br/>async context manager<br/>config + hooks + triggers + tools + chat()"]
    end
    subgraph L2["Layer 2 - Session"]
        Conv["Conversation<br/>history, turns, compaction,<br/>usage, send()/receive_steps()/chat()"]
        CR["ChatResponse<br/>async stream of chunks<br/>text(), thoughts, tool_calls"]
        TR["ToolRunner<br/>registry + executor<br/>sync/async tools"]
        HR["HookRunner<br/>dispatch engine<br/>9 hook types"]
        TGR["TriggerRunner<br/>background tasks<br/>asyncio tasks"]
    end
    subgraph L3["Layer 3 - Adapter"]
        Conn["Connection<br/>send(), receive_steps(),<br/>disconnect(), cancel()"]
        CS["ConnectionStrategy<br/>process mgmt, transport,<br/>auth, health check"]
    end

    Agent --> Conv
    Agent --> TR
    Agent --> HR
    Agent --> TRG
    Conv --> Conn
    Conn --> CS
    TR -.-> Conn
    HR -.-> Conn
    TGR -.-> Conn

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, ToolRunner, HookRunner, TriggerRunner
Layer 3 -- Adapter	Transport and backend abstraction	Connection, ConnectionStrategy, LocalConnection

---

Layer 1: Agent

The Agent class manages the full lifecycle: binary discovery, tool wiring, hook registration, policy defaults, and connection startup -- all behind a single async context manager.

import asyncio
from google.antigravity import Agent, LocalAgentConfig

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

asyncio.run(main())

Agent Lifecycle

class Agent:
    def __init__(self, config: AgentConfig):
        # Deep-copies config, stores pending hooks/triggers

    async def __aenter__(self) -> "Agent":
        # 1. Create HookRunner, register pending hooks
        # 2. Apply policies (safety guard: write tools need policies)
        # 3. Create ToolRunner with registered tools
        # 4. Create ConnectionStrategy from config
        # 5. Enter Conversation.create(strategy)
        # 6. Start TriggerRunner if triggers exist
        # 7. Wire ToolContext into ToolRunner

    async def __aexit__(self, ...):
        # Delegates to AsyncExitStack (stops triggers, disconnects, etc.)

    async def chat(self, prompt: Content) -> ChatResponse:
        # Delegates to self.conversation.chat(prompt)

    def register_hook(self, hook: Hook): ...
    def register_trigger(self, trigger: Trigger): ...

Safety Guard

The Agent enforces a critical safety guard at startup: if write tools or MCP servers are enabled but no policies and no PreToolCallDecideHook are registered, it raises ValueError. This prevents accidentally deploying an unguarded agent with destructive capabilities.

# This raises ValueError:
config = LocalAgentConfig(
    capabilities=CapabilitiesConfig(),  # enables all tools
    policies=[],                        # no safety policies
)
# Fix: add policies
config = LocalAgentConfig(
    capabilities=CapabilitiesConfig(),
    policies=[policy.deny("*"), policy.allow("view_file")],
)

---

Core Types (types.py)

All public SDK interfaces use canonical Pydantic V2 boundary types. No proto dependencies at this layer.

Configuration Types

DEFAULT_MODEL = "gemini-3.5-flash"

class ThinkingLevel(str, enum.Enum):
    MINIMAL = "minimal"
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"

class GenerationConfig(pydantic.BaseModel):
    thinking_level: ThinkingLevel | None = None

class ModelEntry(pydantic.BaseModel):
    name: str
    api_key: str | None = None          # per-model override
    generation: GenerationConfig = ...

class ModelConfig(pydantic.BaseModel):
    default: ModelEntry = ModelEntry(name="gemini-3.5-flash")
    image_generation: ModelEntry = ModelEntry(name="gemini-3.1-flash-image-preview")

class GeminiConfig(pydantic.BaseModel):
    api_key: str | None = None          # falls back to $GEMINI_API_KEY
    vertex: bool = False                # Vertex AI backend
    project: str | None = None          # GCP project
    location: str | None = None         # GCP region
    models: ModelConfig = ...

System Instructions

Two modes -- replacement vs. augmentation:

class CustomSystemInstructions(pydantic.BaseModel):
    """Full replacement. Use with caution."""
    text: str

class TemplatedSystemInstructions(pydantic.BaseModel):
    """Append to defaults (recommended)."""
    identity: str | None = None
    sections: list[SystemInstructionSection] = []

SystemInstructions = CustomSystemInstructions | TemplatedSystemInstructions

Builtin Tools

class BuiltinTools(str, enum.Enum):
    LIST_DIR = "list_directory"
    SEARCH_DIR = "search_directory"
    FIND_FILE = "find_file"
    VIEW_FILE = "view_file"
    CREATE_FILE = "create_file"
    EDIT_FILE = "edit_file"
    RUN_COMMAND = "run_command"
    ASK_QUESTION = "ask_question"
    START_SUBAGENT = "start_subagent"
    GENERATE_IMAGE = "generate_image"
    FINISH = "finish"

    @classmethod
    def read_only(cls) -> list["BuiltinTools"]:
        return [cls.LIST_DIR, cls.SEARCH_DIR, cls.FIND_FILE, cls.VIEW_FILE, cls.FINISH]

    @classmethod
    def file_tools(cls) -> list["BuiltinTools"]:
        return [cls.VIEW_FILE, cls.CREATE_FILE, cls.EDIT_FILE]

CapabilitiesConfig

Controls which tools the harness exposes to the model:

class CapabilitiesConfig(pydantic.BaseModel):
    enable_subagents: bool = True
    enabled_tools: list[BuiltinTools] | None = None    # allowlist
    disabled_tools: list[BuiltinTools] | None = None   # denylist (mutually exclusive)
    compaction_threshold: int | None = None
    image_model: str = "gemini-3.1-flash-image-preview"
    finish_tool_schema_json: str | None = None

MCP Server Configs

class McpStdioServer(BaseMcpServerConfig):
    command: str
    args: list[str] = []
    enabled_tools: list[str] | None = None
    disabled_tools: list[str] | None = None

class McpStreamableHttpServer(BaseMcpServerConfig):
    url: str
    headers: dict[str, str] | None = None
    timeout: float = 30.0
    sse_read_timeout: float = 300.0

McpServerConfig = McpStdioServer | McpStreamableHttpServer

Step Model

Every agent action produces a Step:

class Step(pydantic.BaseModel):
    id: str = ""
    step_index: int = 0
    type: StepType           # TEXT_RESPONSE | TOOL_CALL | SYSTEM_MESSAGE | COMPACTION | FINISH
    source: StepSource       # SYSTEM | USER | MODEL
    target: StepTarget       # USER | ENVIRONMENT
    status: StepStatus       # ACTIVE | DONE | WAITING_FOR_USER | ERROR | CANCELED
    content: str = ""
    content_delta: str = ""
    thinking: str = ""
    thinking_delta: str = ""
    tool_calls: list[ToolCall] = []
    error: str = ""
    is_complete_response: bool | None = None
    structured_output: Any | None = None
    usage_metadata: UsageMetadata | None = None

Streaming Types

class StreamChunk(pydantic.BaseModel):
    step_index: int

class Thought(StreamChunk):
    text: str
    signature: bytes | None = None

class Text(StreamChunk):
    text: str

Multimodal Content

# Supported MIME types
SUPPORTED_IMAGE_MIMES = {"image/bmp", "image/jpeg", "image/png", "image/webp"}
SUPPORTED_DOCUMENT_MIMES = {"application/pdf", "application/json", "text/plain", ...}
SUPPORTED_AUDIO_MIMES = {"audio/wav", "audio/mp3", "audio/ogg", ...}
SUPPORTED_VIDEO_MIMES = {"video/mp4", "video/webm", "video/quicktime", ...}

# Content primitives
Image(_BaseMedia)       # data: bytes, mime_type: str, description: str | None
Document(_BaseMedia)
Audio(_BaseMedia)
Video(_BaseMedia)

ContentPrimitive = str | Image | Document | Audio | Video | SlashCommand
Content = ContentPrimitive | Sequence[ContentPrimitive]

# Auto-resolve from file path
from google.antigravity.types import from_file
doc = from_file("spec.pdf")  # Returns Document based on MIME guess

Error Hierarchy

AntigravityConnectionError(Exception)     # Fatal protocol-level errors
AntigravityCancelledError(CancelledError) # Programmatic cancellation
AntigravityExecutionError(Exception)      # Terminal agent loop errors
AntigravityValidationError(Exception)     # Wraps Pydantic ValidationError

---

Layer 2: Conversation

The Conversation class wraps a Connection with session state management.

from google.antigravity.connections.local import LocalConnectionStrategy
from google.antigravity.conversation.conversation import Conversation

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}")

    # 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)

ChatResponse

Returned by conversation.chat() and agent.chat(). An async stream with independent cursors over a shared buffer.

response = await agent.chat("Write a poem.")

# 1. Stream text tokens directly
async for token in response:
    sys.stdout.write(token)

# 2. Stream reasoning/thinking
async for thought in response.thoughts:
    show_thinking_bubble(thought)

# 3. Stream tool call dispatches
async for call in response.tool_calls:
    show_executing_spinner(call.name)

# 4. Resolve everything
full_text = await response.text()
all_chunks = await response.resolve()

# 5. Cancel mid-stream
await response.cancel()

# 6. Token usage
usage = response.usage_metadata

Every iterator returns an independent cursor over the shared buffer. Cursors are safe to consume sequentially or concurrently via asyncio.gather.

History Management

conversation.history            # Full step list
conversation.turn_count         # Number of send() calls
conversation.compaction_indices # Where context was compacted
conversation.total_usage        # Cumulative UsageMetadata
conversation.last_turn_usage    # UsageMetadata for current turn
conversation.clear_history()    # Free memory, keep session alive

---

Tool System

ToolRunner

Registry and executor for in-process Python tools. Supports both sync and async callables.

from google.antigravity.tools.tool_runner import ToolRunner

def get_weather(city: str) -> str:
    """Returns the current weather for a city."""
    return f"It's sunny in {city}."

tool_runner = ToolRunner(tools=[get_weather])

# Execute by name
result = await tool_runner.execute("get_weather", city="Tokyo")

# Batch process ToolCall objects
from google.antigravity import types
calls = [types.ToolCall(name="get_weather", args={"city": "Osaka"})]
results = await tool_runner.process_tool_calls(calls)

Key behaviors:

• Sync tools run in a separate thread via asyncio.to_thread() to avoid blocking the event loop
• process_tool_calls() executes calls concurrently via asyncio.gather()
• Unknown tools return ToolResult(error="Unknown tool") instead of raising
• ToolWithSchema wrapper for tools with explicit JSON Schema (e.g., MCP server tools)

ToolContext

Conversation-aware context injected into tools that declare a ToolContext-typed parameter. Provides access to conversation capabilities and a per-session key-value store.

from google.antigravity.tools.tool_context import ToolContext

def record_fruit(sku: str, count: int, ctx: ToolContext) -> str:
    """Records fruit count. ctx is injected automatically."""
    current = ctx.get_state("fruit_counts", {})
    current[sku] = current.get(sku, 0) + count
    ctx.set_state("fruit_counts", current)
    return f"Total for {sku}: {current[sku]}"

# ctx also exposes:
# ctx.conversation_id    # Session identifier
# ctx.is_idle            # Connection idle state
# await ctx.send(msg)    # Push message into conversation

At registration time, the ToolRunner inspects each tool's signature for a ToolContext-typed parameter and caches the result. At execution time, the context is injected automatically. Schema generation strips the parameter so the model never sees it.

---

Hook System

Hook Taxonomy

Three categories with strict semantics:

Category	Purpose	Blocks?	Modifies Data?	Examples
InspectHook	Observability, logging	No	No	PostToolCallHook, OnSessionStartHook
DecideHook	Policy enforcement	Yes	No	PreToolCallDecideHook, PreTurnHook
TransformHook	Data transformation, recovery	Yes	Yes	OnToolErrorHook, OnInteractionHook

Hook Interfaces

# Session lifecycle
class OnSessionStartHook(InspectHook[None]): ...
class OnSessionEndHook(InspectHook[None]): ...

# Turn lifecycle
class PreTurnHook(DecideHook[types.Content]): ...     # data = user prompt
class PostTurnHook(InspectHook[str]): ...              # data = response text

# Tool lifecycle
class PreToolCallDecideHook(DecideHook[types.ToolCall]): ...   # data = ToolCall
class PostToolCallHook(InspectHook[types.ToolResult]): ...     # data = ToolResult
class OnToolErrorHook(TransformHook[Exception, Any]): ...      # returns recovery value

# Interaction
class OnInteractionHook(TransformHook[AskQuestionInteractionSpec, QuestionHookResult]): ...

# Compaction
class OnCompactionHook(InspectHook): ...

Decorator Factories

from google.antigravity.hooks import hooks

@hooks.pre_turn
async def my_pre_turn(data):
    return hooks.HookResult(allow=True)

@hooks.post_tool_call
async def my_post_tool(data):
    print(f"Tool {data.name} completed")

@hooks.on_tool_error
async def my_error_recovery(error):
    return f"Recovered from: {error}"

@hooks.on_session_start
async def on_start():
    print("Session started")

Context Hierarchy

SessionContext (session-wide)
  └── TurnContext (per turn)
        └── OperationContext (per tool call / operation)

State flows downward: SessionContext.get() is visible to TurnContext and OperationContext, but not vice versa. Note: HookContext and ToolContext are independent -- they do not share state.

HookRunner Dispatch

The HookRunner manages all registered hooks and dispatches events in strict order:

class HookRunner:
    # Registration
    def register_hook(self, hook: Any) -> None: ...

    # Dispatch methods (called by Connection layer)
    async def dispatch_session_start(self) -> None: ...
    async def dispatch_session_end(self) -> None: ...
    async def dispatch_pre_turn(self, prompt) -> tuple[HookResult, TurnContext]: ...
    async def dispatch_post_turn(self, turn_context, response) -> None: ...
    async def dispatch_pre_tool_call(self, turn_context, tool_call) -> tuple[HookResult, ToolCall, OperationContext]: ...
    async def dispatch_post_tool_call(self, op_context, result) -> None: ...
    async def dispatch_on_tool_error(self, op_context, error) -> tuple[HookResult, Any]: ...
    async def dispatch_interaction(self, turn_context, spec) -> tuple[HookResult, Any, OperationContext]: ...
    async def dispatch_compaction(self, turn_context, data) -> None: ...

Dispatch order for tool calls: PreToolCallDecideHook -> (tool execution) -> PostToolCallHook or OnToolErrorHook.

---

Policy System

Declarative API for tool call policies. Policies are evaluated using a priority-based model where specificity and safety determine precedence.

Priority Buckets (lower = higher priority)

Level	Specificity	Decision	Example
0	Specific	DENY	deny("run_command")
1	Specific	ASK_USER	ask_user("run_command", handler=fn)
2	Specific	APPROVE	allow("run_command")
3	Prefix wildcard	DENY	deny(mcp_server, ["tool"])
4	Prefix wildcard	ASK_USER	ask_user(mcp_server, ["tool"], handler=fn)
5	Prefix wildcard	APPROVE	allow(mcp_server, ["tool"])
6	Global wildcard	DENY	deny("*")
7	Global wildcard	ASK_USER	ask_user("*", handler=fn)
8	Global wildcard	APPROVE	allow("*")

Within each bucket, first match wins (short-circuit).

Policy Builders

from google.antigravity.hooks import policy

# Simple policies
policy.deny("*")                              # Block everything
policy.allow("view_file")                     # Allow one tool
policy.deny_all()                             # Alias for deny("*")
policy.allow_all()                            # Alias for allow("*")

# Conditional predicates
policy.deny("run_command",
    when=lambda args: "rm" in args.get("CommandLine", ""))

# User confirmation
policy.ask_user("run_command", handler=my_approval_fn)

# Safe defaults (read-only + ask for everything else)
policy.safe_defaults(handler=my_approval_fn)

# Default LocalAgentConfig policy
policy.confirm_run_command()  # denies run_command, allows everything else

# Workspace scoping
policy.workspace_only(["/path/to/workspace"])

# MCP tool policies
policy.allow(mcp_server_config, ["tool1", "tool2"])
policy.deny(mcp_server_config)  # all tools on server

Enforcement

from google.antigravity.hooks import policy

policies = [
    policy.deny("*"),
    policy.allow("view_file"),
    policy.ask_user("run_command", handler=my_handler),
]

hook = policy.enforce(policies, mcp_servers=[...])
# Returns a PreToolCallDecideHook ready for registration

Disabling vs. Denying Tools

Mechanism	Model sees tool?	Token cost	Best for
CapabilitiesConfig.disabled_tools	No	None	Tools irrelevant to agent purpose
policy.deny()	Yes	Wasted on failed calls	Conditional/argument-dependent restrictions

---

Trigger System

Triggers are long-lived async functions that run alongside the agent session, reacting to external events.

Core Types

class TriggerContext:
    async def send(self, content: str) -> None:
        """Push a message into the agent conversation."""

# A Trigger is any async def accepting TriggerContext
Trigger = Callable[[TriggerContext], Awaitable[None]]

Helper Factories

from google.antigravity.triggers import every, on_file_change

# Fixed interval (first invocation after first interval, not immediate)
my_trigger = every(300, check_status)  # Every 5 minutes

# File system watching (requires `watchfiles` package)
config_watcher = on_file_change("/etc/app/config.yaml", handle_change)
# Callback receives (ctx, list[FileChange])
# FileChange.kind: ADDED | MODIFIED | DEDED
# FileChange.path: absolute path

TriggerRunner

class TriggerRunner:
    async def start(self) -> None: ...   # Creates asyncio.Task per trigger
    async def stop(self) -> None: ...    # Cancels all tasks
    # Also supports async with:
    async with TriggerRunner(triggers=[...], connection=conn) as runner:
        ...

• Each trigger runs as an independent asyncio task
• Unhandled exceptions are logged but do not crash the session
• No auto-restart, no ordering guarantees

---

Connection Abstraction

ABCs

class AgentConfig(ABC, pydantic.BaseModel):
    system_instructions: str | SystemInstructions | None = None
    capabilities: CapabilitiesConfig = ...
    tools: list[Callable] = []
    policies: list[Any] = []
    hooks: list[Any] = []
    triggers: list[Any] = []
    mcp_servers: list[McpServerConfig] = []
    workspaces: list[str] = []
    conversation_id: str | None = None
    save_dir: str | None = None
    response_schema: dict | BaseModel | str | None = None
    skills_paths: list[str] = []

    @abstractmethod
    def create_strategy(self, *, tool_runner, hook_runner) -> ConnectionStrategy: ...

class Connection(ABC):
    @property
    def is_idle(self) -> bool: ...
    @property
    def conversation_id(self) -> str: ...

    @abstractmethod
    async def send(self, prompt: Content | None, **kwargs) -> None: ...
    @abstractmethod
    def receive_steps(self) -> AsyncIterator[Step]: ...

    async def disconnect(self) -> None: ...
    async def cancel(self) -> None: ...
    async def send_tool_results(self, results: list[ToolResult]) -> None: ...
    async def send_trigger_notification(self, content: str) -> None: ...
    async def wait_for_idle(self) -> None: ...

class ConnectionStrategy(ABC):
    @abstractmethod
    def connect(self) -> Connection: ...
    @abstractmethod
    async def __aenter__(self) -> None: ...
    @abstractmethod
    async def __aexit__(self, ...) -> None: ...

LocalConnection (Layer 3)

The primary (and currently only) connection implementation. Communicates with a Go-based local harness via WebSocket.

Startup sequence:

1. Validate API key (Gemini or Vertex AI)
2. Build HarnessConfig protobuf from SDK config objects
3. Spawn localharness binary as subprocess
4. Exchange InputConfig / OutputConfig via stdin/stdout (length-prefixed protobuf)
5. Connect to harness WebSocket (with retry backoff)
6. Send InitializeConversationEvent with full config
7. Dispatch on_session_start hooks

Wire protocol: JSON-serialized protobuf messages over WebSocket:

• InputEvent (SDK -> harness): user_input, tool_response, tool_confirmation, question_response, halt_request, automated_trigger
• OutputEvent (harness -> SDK): step_update, trajectory_state_update, tool_call

Tool execution flow:

1. Harness sends tool_call event
2. LocalConnection._handle_tool_call() dispatches PreToolCallDecideHook
3. If allowed, ToolRunner.process_tool_calls() executes the tool
4. PostToolCallHook or OnToolErrorHook dispatched
5. ToolResponse sent back via WebSocket

Subagent tracking:

• cascade_id identifies the parent trajectory
• Steps from subagent trajectories (different trajectory_id) are tracked
• Connection is idle only when parent AND all subagents complete
• Subagent final responses are captured for PostToolCallHook

LocalAgentConfig

class LocalAgentConfig(AgentConfig):
    # Defaults
    capabilities: CapabilitiesConfig = CapabilitiesConfig()  # all tools enabled
    policies: list = policy.confirm_run_command              # denies run_command
    workspaces: list = [os.getcwd()]                         # current directory

    # Shorthand fields (flow into gemini_config)
    model: str | None = None          # e.g. "gemini-2.5-pro"
    api_key: str | None = None
    vertex: bool | None = None
    project: str | None = None
    location: str | None = None

    # Validators
    # - model/api_key shorthand conflicts with gemini_config raise ValueError
    # - workspaces auto-prepend workspace_only() policies
    # - app_data_dir defaults to ~/.gemini/antigravity

---

Interactive Utilities

from google.antigravity.utils.interactive import run_interactive_loop

async with Agent(config) as agent:
    await run_interactive_loop(agent)

Features:

• async_input() -- async stdin reader (daemon thread, not asyncio.to_thread)
• Spinner -- terminal spinner with TTY detection
• ToolConfirmationHook -- prompts y/n before tool execution
• AskQuestionHook -- handles agent question requests
• ask_user_handler -- policy handler for ASK_USER
• Automatic upgrade of confirm_run_command deny to ask_user for interactive sessions

---

Multimodal Input

from google.antigravity import Agent, LocalAgentConfig
from google.antigravity.types import Image, from_file

config = LocalAgentConfig(system_instructions="You are an expert architect.")
async with Agent(config) as agent:
    # From file path
    spec = from_file("spec.pdf")

    # From raw bytes
    chart = Image(data=b"raw_png_bytes", mime_type="image/png", description="Blueprint")

    # Mixed prompt
    response = await agent.chat([
        "Analyze this chart against the spec:",
        chart,
        spec,
    ])
    print(await response.text())

---

Structured Output

from pydantic import BaseModel

class AnalysisResult(BaseModel):
    summary: str
    risk_level: str
    recommendations: list[str]

config = LocalAgentConfig(
    response_schema=AnalysisResult,
)
async with Agent(config) as agent:
    response = await agent.chat("Analyze the security posture of this codebase.")
    structured = await response.structured_output()
    # structured is a parsed dict matching AnalysisResult schema

The schema is serialized to JSON and passed as finish_tool_schema_json in CapabilitiesConfig, enabling the harness to extract structured output from the FINISH step.

---

Deep-Dive Pages

For detailed analysis of specific subsystems, see the dedicated deep-dive pages:

Topic	Deep-Dive Page	Key Content
Localharness binary	sdk-test-localharness	Go binary subprocess model, WebSocket protocol, 7-phase lifecycle, binary discovery chain
Protobuf wire format	sdk-test-protobuf	Every message type, all fields, oneof patterns, JSON serialization rules
Hook system	sdk-test-hooks-deep	All 9 hook types, Inspect/Decide/Transform categories, HookRunner dispatch, 9-level policy system
Authentication	sdk-test-auth-chain	TrySilentAuth flow, keychain format, token lifecycle, PKCE flow
gRPC protocol	sdk-test-grpc-protocol	Service methods, authentication headers, streaming behavior, error recovery
Decompiled patterns	sdk-test-decompiled-patterns	10 architectural patterns from 3305 decompiled functions
DeepSeek + Heartbeat	sdk-test-deepseek-heartbeat	DeepSeek V4 integration, Heartbeat Protocol, Proto-Unicode sequences

Cross-References

• sdk-test-architecture — JETSKY ecosystem architecture, auth chain, token lifecycle, gRPC protocol
• sdk-test-reverse-engineering — Full RE analysis of the agy binary (3305 decompiled functions)
• sdk-test-integrations — DeepSeek, heartbeat, Gemma 4, SDK integration details
• antigravity-2.0 -- The broader Antigravity research harness ecosystem
• antigravity-ecosystem -- How the SDK fits with Antigravity IDE and CLI
• mcp-ecosystem -- Model Context Protocol integration details

---

Source: antigravity-sdk-python v0.1.2, Apache 2.0, Google LLC.