模块 3：使用 SDK 构建 AGY 代理¶

from google.antigravity import Agent, LocalAgentConfig
from google.antigravity.hooks import policy
print("google-antigravity installed ✅")

def get_file_contents(file_path: str) -> str:
    """Read and return the contents of a file at the given path.

    Args:
        file_path: Absolute or relative path to the file.

    Returns:
        The file contents as a string, or an error message if not found.
    """
    try:
        with open(file_path, "r", encoding="utf-8") as f:
            return f.read()
    except FileNotFoundError:
        return f"Error: File not found at {file_path}"

from google.antigravity.tools.tool_context import ToolContext

def record_finding(
    severity: str,
    message: str,
    ctx: ToolContext,
) -> dict:
    """Records a review finding into session state.

    Args:
        severity: One of 'critical', 'warning', 'info'.
        message: Description of the finding.

    Returns:
        Confirmation dict with the finding index.
    """
    findings = ctx.get_state("findings", [])
    findings.append({"severity": severity, "message": message})
    ctx.set_state("findings", findings)
    return {"status": "recorded", "index": len(findings) - 1}

import asyncio
from google.antigravity import Agent, LocalAgentConfig
from google.antigravity.hooks import policy

config = LocalAgentConfig(
    model="gemini-3.5-flash",
    system_instructions="""You are a code reviewer specialising in Python.
When given a file path, read the file and provide a structured review covering:
- Correctness and edge cases
- Code style and readability
- Security concerns
- Suggested improvements

Always read the file first before commenting. Be specific — cite line numbers.
""",
    tools=[get_file_contents, record_finding],
    policies=[policy.allow_all()],          # autonomous — no interactive prompts
    workspaces=["/path/to/project"],        # file ops scoped to this directory
)

async def main():
    async with Agent(config) as agent:
        response = await agent.chat("Review src/auth/login.py")
        print(await response.text())

asyncio.run(main())

from pathlib import Path
import re

def load_skill(skill_name: str) -> str:
    """Load a SKILL.md and strip YAML frontmatter."""
    skill_path = Path("skills") / skill_name / "SKILL.md"
    if skill_path.exists():
        content = skill_path.read_text(encoding="utf-8")
        return re.sub(r'^---\n.*?---\n', '', content, flags=re.DOTALL).strip()
    return ""

review_guidelines = load_skill("python-review")

config = LocalAgentConfig(
    model="gemini-3.5-flash",
    system_instructions=f"""You are a code reviewer.

## Review Guidelines
{review_guidelines}
""",
    tools=[get_file_contents],
    policies=[policy.allow_all()],
)

from google.antigravity.hooks import policy

# Fully autonomous — approve all tool calls (use for trusted, sandboxed agents)
policies=[policy.allow_all()]

# Default behaviour — ask user before running shell commands, allow everything else
policies=[policy.confirm_run_command()]

# Fine-grained rules (evaluated in order, first match wins)
async def approval_handler(tool_call) -> bool:
    answer = input(f"Allow {tool_call.name}? [y/N]: ")
    return answer.lower() == "y"

policies=[
    policy.deny("run_command"),                 # never run shell commands
    policy.allow("view_file"),                  # always allow reading
    policy.ask_user("edit_file", handler=approval_handler),  # ask before every write
    policy.allow("*"),                          # allow everything else
]

# Conditional deny — block dangerous patterns
policy.deny("run_command", when=lambda args: "rm -rf" in args.get("CommandLine", ""))

# Scope file operations to a specific directory
policy.workspace_only(["/path/to/project"])

from google.antigravity.hooks import hooks
from google.antigravity.types import ToolCall, ToolResult, HookResult

# Block dangerous tool calls BEFORE they execute
@hooks.pre_tool_call_decide
async def security_guard(tool_call: ToolCall) -> HookResult:
    if tool_call.name == "run_command":
        cmd = tool_call.args.get("CommandLine", "")
        if any(danger in cmd for danger in ["rm -rf", "drop table", "DELETE FROM"]):
            return HookResult(allow=False, message=f"Blocked dangerous command: {cmd}")
    return HookResult(allow=True)

# Log all tool completions (non-blocking, read-only)
@hooks.post_tool_call
async def audit_logger(tool_result: ToolResult) -> None:
    print(f"[AUDIT] tool={tool_result.name} success={tool_result.success}")

# Initialise state when a session begins
@hooks.on_session_start
async def initialise_state() -> None:
    print("[AGENT] Session started — ready.")

config = LocalAgentConfig(
    hooks=[security_guard, audit_logger, initialise_state],
    policies=[policy.allow_all()],
    model="gemini-3.5-flash",
    system_instructions="You are a code reviewer.",
    tools=[get_file_contents],
)

from google.antigravity.types import BuiltinTools, CapabilitiesConfig

config = LocalAgentConfig(
    capabilities=CapabilitiesConfig(
        enable_subagents=True,
        enabled_tools=BuiltinTools.all_tools(),
    ),
    policies=[policy.allow_all()],
    model="gemini-3.1-pro-preview",
    system_instructions="""You are an engineering lead.
For complex tasks, spawn focused subagents to handle each part in parallel.
Synthesise their outputs into a final summary.""",
)

async def sequential_review(file_path: str):
    # Step 1 — read and summarise the file
    async with Agent(reader_config) as reader:
        r1 = await reader.chat(f"Read and summarise {file_path}")
        summary = await r1.text()

    # Step 2 — security audit using the summary
    async with Agent(security_config) as auditor:
        r2 = await auditor.chat(f"Security audit this code summary:\n\n{summary}")
        report = await r2.text()

    return report

async def parallel_analysis(file_path: str):
    async with (
        Agent(security_config) as security_agent,
        Agent(style_config)    as style_agent,
        Agent(perf_config)     as perf_agent,
    ):
        results = await asyncio.gather(
            security_agent.chat(f"Security review: {file_path}"),
            style_agent.chat(f"Style review: {file_path}"),
            perf_agent.chat(f"Performance review: {file_path}"),
        )
        texts = await asyncio.gather(*[r.text() for r in results])

    return {
        "security": texts[0],
        "style":    texts[1],
        "perf":     texts[2],
    }

async with Agent(config) as agent:
    response = await agent.chat("Write a detailed security report...")

    # Stream text deltas as they arrive
    async for delta in response:
        print(delta, end="", flush=True)

    # Stream reasoning/thinking (if thinking enabled)
    async for thought in response.thoughts:
        print(f"[thinking] {thought}")

import asyncio
import pydantic
from google.antigravity import Agent, LocalAgentConfig
from google.antigravity.hooks import policy

class ReviewResult(pydantic.BaseModel):
    issues: list[str]
    severity: str          # 'critical' | 'warning' | 'info'
    recommendation: str

config = LocalAgentConfig(
    response_schema=ReviewResult,
    system_instructions="Analyse the code and return structured output via the finish tool.",
    policies=[policy.allow_all()],
    model="gemini-3.5-flash",
    tools=[get_file_contents],
)

async def main():
    async with Agent(config) as agent:
        response = await agent.chat("Review src/auth/login.py")
        result = await response.structured_output()   # dict matching ReviewResult schema
        print(result["severity"], result["issues"])

asyncio.run(main())

# First session — save the conversation ID
async with Agent(config) as agent:
    await agent.chat("Analyse this codebase and build a mental model.")
    conv_id = agent.conversation_id   # persist this

# Later session — resume exactly where you left off
resume_config = LocalAgentConfig(
    conversation_id=conv_id,
    save_dir="/path/where/first/session/was/saved",
    model="gemini-3.5-flash",
    policies=[policy.allow_all()],
)
async with Agent(resume_config) as agent:
    await agent.chat("Now suggest the top 3 refactoring priorities.")

import asyncio
from google.antigravity import Agent, LocalAgentConfig
from google.antigravity.hooks import policy
from google.antigravity.triggers import every, on_file_change, TriggerContext

# Poll every 60 seconds
async def check_for_new_issues(ctx: TriggerContext) -> None:
    await ctx.send("Scan the repo for any new TODO comments added since last run.")

# React to file changes
async def on_code_change(ctx: TriggerContext, changes) -> None:
    paths = [c.path for c in changes]
    await ctx.send(f"Files changed: {paths}. Run quick security check.")

config = LocalAgentConfig(
    triggers=[
        every(60.0, check_for_new_issues),
        on_file_change("/path/to/src", on_code_change),
    ],
    policies=[policy.allow_all()],
    model="gemini-3.5-flash",
    system_instructions="You are a background code monitor.",
)

async def main():
    # The agent runs indefinitely, responding to triggers
    async with Agent(config) as agent:
        await asyncio.Event().wait()  # keep alive

asyncio.run(main())