--- name: sdk-test-antigravity-sdk type: reference title: "Google Antigravity Python SDK" description: "Complete reference for the Antigravity 2.0 Python SDK: three-layer architecture, Agent API, tool/hook/trigger systems, connection abstraction, and protobuf wire protocol" tags: [antigravity, sdk, python, agent-framework, gemini, mcp, hooks, triggers, tools] timestamp: 2026-07-21 --- # 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 ```toml # 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 ```mermaid graph TB subgraph L1["Layer 1 - Simplified"] Agent["Agent
async context manager
config + hooks + triggers + tools + chat()"] end subgraph L2["Layer 2 - Session"] Conv["Conversation
history, turns, compaction,
usage, send()/receive_steps()/chat()"] CR["ChatResponse
async stream of chunks
text(), thoughts, tool_calls"] TR["ToolRunner
registry + executor
sync/async tools"] HR["HookRunner
dispatch engine
9 hook types"] TGR["TriggerRunner
background tasks
asyncio tasks"] end subgraph L3["Layer 3 - Adapter"] Conn["Connection
send(), receive_steps(),
disconnect(), cancel()"] CS["ConnectionStrategy
process mgmt, transport,
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. ```python 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 ```python 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. ```python # 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 ```python 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: ```python 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 ```python 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: ```python 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 ```python 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`: ```python 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 ```python class StreamChunk(pydantic.BaseModel): step_index: int class Thought(StreamChunk): text: str signature: bytes | None = None class Text(StreamChunk): text: str ``` ### Multimodal Content ```python # 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 ```python 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. ```python 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. ```python 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 ```python 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. ```python 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. ```python 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 ```python # 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 ```python 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: ```python 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 ```python 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 ```python 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 ```python 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 ```python 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 ```python 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 ```python 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 ```python 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 ```python 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 ```python 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 ```python 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.*