Antigravity SDK 심화편—Managed Agents API·GCP 엔터프라이즈 연동·CI/CD 파이프라인 실전 구축: Antigravity 2.0

 

기존 글에서 SDK 기초는 다뤘습니다. 이번엔 그 다음 단계입니다. Managed Agents API로 코드 한 줄로 격리 샌드박스를 띄우고, GCP와 연결하고, GitHub Actions에 박아 넣는 것까지.

---

📌 핵심 요약
→ Antigravity SDK = 에이전트 하네스를 내 코드에 임베딩하는 Python/JS 라이브러리
→ Managed Agents API: 목표만 선언 → Google이 샌드박스·루프·재시도 전부 관리
→ 과금 구조: 토큰 단위가 아닌 실행(run) 단위 — 장기 작업에 훨씬 유리
→ Agent 클래스: 바이너리 탐색·툴 와이어링·훅 등록·정책 기본값 전부 자동 관리
→ GCP Enterprise: SLA 보장·VPC 프라이빗 실행·IAM·감사 로그 — 엔터프라이즈 필수
→ GitHub Actions 통합: PR마다 에이전트가 코드 리뷰·보안 스캔·문서 업데이트 자동 실행
→ MCP 서버 연동: Antigravity 에이전트가 GitHub·BigQuery·Cloud Run 직접 조작
→ 자격증명 노출 주의: API 키를 모델 컨텍스트에 직접 전달하면 유출 위험

---

실전1 — SDK 설치와 3계층 아키텍처 이해

Antigravity SDK는 에이전트 루프·바이너리 탐색·툴 와이어링·훅 등록·정책 기본값을 추상화 계층 뒤에 숨겨 개발자가 에이전트가 어떻게 동작하는지가 아니라 무엇을 하는지에 집중할 수 있게 합니다.

# 설치 — PyPI에서 플랫폼별 바이너리 포함 wheel 자동 설치
pip install google-antigravity

# 주의: GitHub 레포 클론만으로는 동작 안 함
# 반드시 PyPI에서 설치해야 컴파일된 런타임 바이너리 포함됨

# API 키 설정
export GEMINI_API_KEY="your_gemini_api_key"

# 설치 확인
python -c "from google.antigravity import Agent; print('✅ 설치 완료')"

[SDK 3계층 아키텍처]

Layer 1 — Agent & Config
  Agent 클래스:      전체 에이전트 라이프사이클 관리
  AgentConfig:      클라우드 실행 설정 (Managed Agents)
  LocalAgentConfig: 로컬 실행 설정 (자체 인프라)

Layer 2 — Conversation & Streaming
  ChatResponse:     단일 응답 처리
  Streaming API:    토큰·추론 CoT 실시간 스트리밍

Layer 3 — Tools & MCP
  ToolRunner:       커스텀 툴 정의·실행
  McpBridge:        외부 MCP 서버 연결
  HookRunner:       이벤트 기반 훅 실행

→ 일반 개발자: Agent 클래스만 써도 충분
→ 프로덕션 파이프라인: ToolRunner + McpBridge 조합
→ 엔터프라이즈: GCP Enterprise Agent Platform + LocalAgentConfig

---

실전2 — Managed Agents API: 코드 한 줄로 격리 샌드박스

Managed Agents는 일반 모델 호출과 다릅니다. 격리된 Linux 샌드박스, 코드 런타임, 파일시스템, 브라우저 전체를 Google이 호스팅·관리하고 개발자는 API만 씁니다. 토큰 단위가 아닌 실행(run) 단위로 과금됩니다.

# managed_agent_basic.py — Managed Agents API 기본 사용
from google import genai

client = genai.Client()

# 방법 1: genai 클라이언트 직접 사용 (가장 간단)
interaction = client.interactions.create(
    agent="antigravity-preview-05-2026",
    input="Hacker News 상위 10개 기사 요약해서 PDF로 저장해줘",
    environment="remote",   # Google이 관리하는 격리 샌드박스
)

print(interaction.output_text)
# → 에이전트가 웹 브라우징 → 요약 → PDF 저장 → 결과 반환
# → 로컬 환경 오염 없음, 실행 후 샌드박스 자동 폐기

# managed_agent_advanced.py — SDK Agent 클래스로 상세 제어
import asyncio
from google.antigravity import Agent, AgentConfig

async def run_managed_agent():
    config = AgentConfig(
        model="gemini-3.5-flash",
        environment="remote",          # "remote" = Managed, "local" = 자체 인프라
        sandbox=True,                  # 격리 실행
        credential_masking=True,       # API 키 자동 마스킹 (컨텍스트 유출 방지)
        timeout=300,                   # 최대 5분
        tools=[
            "code_execution",          # Python 코드 실행
            "file_management",         # 파일 읽기·쓰기
            "web_browsing",            # 웹 검색·스크래핑
            "shell",                   # bash 명령 실행
        ],
    )

    async with Agent(config=config) as agent:
        # 스트리밍으로 실시간 진행 확인
        async for chunk in agent.stream(
            "우리 레포 src/ 디렉토리 분석해서 "
            "1) 코드 복잡도 높은 파일 TOP 5 "
            "2) 미사용 import 목록 "
            "3) 테스트 없는 함수 목록 "
            "작성하고 code-analysis.md로 저장해줘"
        ):
            if chunk.type == "thought":
                print(f"[추론] {chunk.text}")    # CoT 실시간 출력
            elif chunk.type == "text":
                print(chunk.text, end="", flush=True)

asyncio.run(run_managed_agent())

# 스트리밍 없이 단순 실행 (배치 작업)
async def run_batch_task(task: str) -> str:
    config = AgentConfig(
        model="gemini-3.5-flash",
        environment="remote",
        tools=["code_execution", "file_management"],
        credential_masking=True,
    )

    async with Agent(config=config) as agent:
        response = await agent.chat(task)
        return response.text

# 실행
result = asyncio.run(run_batch_task(
    "package.json 읽고 모든 의존성의 최신 버전 확인해서 "
    "업데이트 필요한 것만 정리한 report.md 작성해줘"
))
print(result)

[Managed Agents 과금 구조]

일반 Gemini API:   입력/출력 토큰 단위 과금
Managed Agents:    실행(run) 단위 + 샌드박스 초당 컴퓨팅

→ 장점: 30분짜리 장기 에이전트 작업도 토큰 폭탄 없음
→ 단점: 짧은 단발성 쿼리는 일반 API가 더 저렴
→ Free tier: 실험용으로 제한적 무료 제공
→ 프로덕션: GCP Enterprise Platform (SLA·VPC·감사 로그)

---

실전3 — 커스텀 툴 + MCP 서버 연동

# custom_tools_mcp.py — 커스텀 툴 정의 + GitHub MCP 연동
import asyncio
from google.antigravity import Agent, LocalAgentConfig
from google.antigravity.tools import ToolRunner, tool

# ── 커스텀 툴 정의 ──────────────────────────────────────
@tool(
    name="get_jira_ticket",
    description="Jira 티켓 상세 정보 조회",
)
async def get_jira_ticket(ticket_id: str) -> dict:
    """에이전트가 Jira 티켓 내용을 읽을 수 있게 됨"""
    import httpx
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"https://company.atlassian.net/rest/api/3/issue/{ticket_id}",
            auth=("user@company.com", JIRA_API_TOKEN),
        )
        data = resp.json()
        return {
            "title":       data["fields"]["summary"],
            "description": data["fields"]["description"],
            "status":      data["fields"]["status"]["name"],
            "assignee":    data["fields"]["assignee"]["displayName"],
        }

@tool(
    name="post_slack_message",
    description="Slack 채널에 메시지 전송",
)
async def post_slack_message(channel: str, message: str) -> str:
    import httpx
    async with httpx.AsyncClient() as client:
        await client.post(
            "https://slack.com/api/chat.postMessage",
            headers={"Authorization": f"Bearer {SLACK_BOT_TOKEN}"},
            json={"channel": channel, "text": message},
        )
    return f"✅ #{channel}에 메시지 전송 완료"

# ── MCP 서버 연결 설정 ───────────────────────────────────
MCP_SERVERS = [
    {
        "name": "github",
        "url": "https://api.githubcopilot.com/mcp/",
        "headers": {"Authorization": f"Bearer {GITHUB_TOKEN}"},
    },
    {
        "name": "bigquery",
        "command": "npx @google/bigquery-mcp-server",
        "env": {"GOOGLE_CLOUD_PROJECT": "my-project"},
    },
]

# ── 에이전트 실행 ────────────────────────────────────────
async def run_pr_automation():
    config = LocalAgentConfig(      # 자체 인프라에서 실행
        model="gemini-3.5-flash",
        tools=ToolRunner(tools=[
            get_jira_ticket,
            post_slack_message,
        ]),
        mcp_servers=MCP_SERVERS,    # GitHub + BigQuery MCP
        credential_masking=True,
        system_instructions="""
        당신은 PR 자동화 에이전트입니다.
        Jira 티켓 내용을 읽고, 코드 변경사항과 비교해서,
        요구사항이 제대로 구현됐는지 검증합니다.
        """,
    )

    async with Agent(config=config) as agent:
        result = await agent.chat(
            "PR #342와 연결된 Jira 티켓 PROJ-1234를 읽고, "
            "PR 변경사항이 티켓 요구사항을 충족하는지 검토해줘. "
            "완료되면 #dev-reviews Slack 채널에 요약 올려줘."
        )
        print(result.text)

asyncio.run(run_pr_automation())

[MCP 서버 연동 패턴]

URL 방식 (원격 MCP):
→ GitHub, Google Cloud, Firebase 등 공식 MCP
→ headers로 인증 토큰 전달
→ credential_masking: True로 컨텍스트 노출 방지

Command 방식 (로컬 MCP):
→ npx, uvx, 직접 바이너리로 실행
→ env로 환경변수 전달
→ 온프레미스 사내 시스템 연동 시 주로 사용

Antigravity 기본 제공 MCP (앱에서 클릭만 하면 연결):
→ GitHub / GitLab / Google Cloud Run / Firebase
→ BigQuery / Kubernetes / Cloud SQL
→ Slack / Jira / Linear

---

실전4 — GCP Enterprise Agent Platform 연동

프로덕션 워크로드를 위한 GCP Agent Platform은 SLA 보장 에이전트 실행, 프라이빗 VPC 및 데이터 주권, Google Cloud Console을 통한 중앙화된 IAM 관리를 제공합니다.

# gcp_enterprise_agent.py — GCP Enterprise 연동
import asyncio
from google.antigravity import Agent, AgentConfig
from google.cloud import secretmanager

# ── Secret Manager에서 자격증명 로드 (코드에 직접 X) ────
def get_secret(secret_id: str) -> str:
    client = secretmanager.SecretManagerServiceClient()
    name = f"projects/my-project/secrets/{secret_id}/versions/latest"
    response = client.access_secret_version(request={"name": name})
    return response.payload.data.decode("UTF-8")

async def run_enterprise_agent(task: str) -> str:
    config = AgentConfig(
        model="gemini-3.5-flash",
        environment="remote",

        # GCP Enterprise 설정
        gcp_project="my-gcp-project",
        gcp_region="us-central1",
        vpc_network="projects/my-project/global/networks/private-vpc",
        # → 에이전트가 프라이빗 VPC 안에서 실행됨
        # → 인터넷 노출 없이 내부 시스템 접근 가능

        service_account="antigravity-agent@my-project.iam.gserviceaccount.com",
        # → 최소 권한 원칙: 이 SA가 접근 가능한 리소스만 에이전트가 사용 가능

        audit_logging=True,         # 모든 에이전트 행동 Cloud Audit Logs 기록
        credential_masking=True,
        timeout=600,                # 엔터프라이즈: 최대 10분

        tools=["code_execution", "file_management", "shell"],
    )

    async with Agent(config=config) as agent:
        response = await agent.chat(task)
        return response.text

# 실행
result = asyncio.run(run_enterprise_agent(
    "BigQuery의 last_30_days 데이터셋에서 "
    "이상 트랜잭션 패턴 분석하고 "
    "Cloud Storage에 anomaly_report_2026.csv로 저장해줘"
))
print(result)

# GCP IAM 최소 권한 설정 — Terraform
resource "google_service_account" "antigravity_agent" {
  account_id   = "antigravity-agent"
  display_name = "Antigravity Agent Service Account"
  project      = var.project_id
}

# 필요한 권한만 부여 (최소 권한 원칙)
resource "google_project_iam_member" "agent_bigquery" {
  project = var.project_id
  role    = "roles/bigquery.dataViewer"    # BigQuery 읽기만
  member  = "serviceAccount:${google_service_account.antigravity_agent.email}"
}

resource "google_project_iam_member" "agent_storage" {
  project = var.project_id
  role    = "roles/storage.objectCreator"  # GCS 쓰기만
  member  = "serviceAccount:${google_service_account.antigravity_agent.email}"
}

# Secret Manager 접근 (API 키 등)
resource "google_secret_manager_secret_iam_member" "agent_secrets" {
  secret_id = google_secret_manager_secret.api_keys.id
  role      = "roles/secretmanager.secretAccessor"
  member    = "serviceAccount:${google_service_account.antigravity_agent.email}"
}

[GCP Enterprise vs Free Tier 차이]

항목               Free Tier              GCP Enterprise
────────────────────────────────────────────────────────
SLA                없음                   99.9% uptime
실행 환경          공용 샌드박스           프라이빗 VPC
데이터 저장        Google 인프라           내 GCP 프로젝트
IAM                없음                   서비스 어카운트
감사 로그          없음                   Cloud Audit Logs
지원               커뮤니티               Google 엔터프라이즈
과금               무료 (제한)            사용량 기반

→ 개인·소규모: Free Tier로 충분
→ 금융·의료·규정 준수 필요: GCP Enterprise 필수
→ 온프레미스 데이터 접근: VPC 피어링 + LocalAgentConfig 조합

---

실전5 — GitHub Actions CI/CD 완전 통합

# .github/workflows/ai-agent-review.yml
name: AI Agent Code Review

on:
  pull_request:
    types: [opened, synchronize]
    paths:
      - 'src/**'
      - 'tests/**'

jobs:
  agent-review:
    runs-on: ubuntu-latest
    permissions:
      contents: read
      pull-requests: write   # PR 코멘트 작성 권한

    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0     # 전체 히스토리 (diff 비교용)

      - name: Python 설정
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - name: Antigravity SDK 설치
        run: pip install google-antigravity

      - name: AI 에이전트 코드 리뷰 실행
        id: agent-review
        env:
          GEMINI_API_KEY: ${{ secrets.GEMINI_API_KEY }}
          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
          PR_NUMBER: ${{ github.event.pull_request.number }}
        run: python .github/scripts/agent_review.py

      - name: 리뷰 결과 PR 코멘트 게시
        uses: actions/github-script@v7
        if: always()
        with:
          script: |
            const fs = require('fs');
            if (!fs.existsSync('review_result.json')) return;
            const result = JSON.parse(fs.readFileSync('review_result.json'));
            const icon = result.passed ? '✅' : '⚠️';
            await github.rest.issues.createComment({
              owner: context.repo.owner,
              repo: context.repo.repo,
              issue_number: context.issue.number,
              body: `## ${icon} AI Agent 코드 리뷰 결과\n\n${result.summary}`
            });

# .github/scripts/agent_review.py
import asyncio, json, os, subprocess
from google.antigravity import Agent, AgentConfig

async def run_pr_review():
    pr_number = os.environ["PR_NUMBER"]

    # PR 변경 파일 목록 가져오기
    diff = subprocess.run(
        ["git", "diff", "origin/main...HEAD", "--name-only"],
        capture_output=True, text=True
    ).stdout.strip()

    # 변경 내용 가져오기
    diff_content = subprocess.run(
        ["git", "diff", "origin/main...HEAD"],
        capture_output=True, text=True
    ).stdout[:15000]   # 토큰 제한으로 앞부분만

    config = AgentConfig(
        model="gemini-3.5-flash",
        environment="remote",
        credential_masking=True,
        timeout=180,
        tools=["code_execution"],   # 코드 분석용
    )

    async with Agent(config=config) as agent:
        response = await agent.chat(f"""
다음 PR #{pr_number} 변경사항을 검토해주세요.

[변경된 파일]
{diff}

[변경 내용 (diff)]
{diff_content}

다음 항목을 검토하고 결과를 JSON으로 출력해주세요:
1. 보안 취약점 (SQL 인젝션, 하드코딩 시크릿, XSS 등)
2. 성능 문제 (N+1 쿼리, 불필요한 루프, 메모리 누수)
3. 코드 품질 (함수 길이, 복잡도, 네이밍)
4. 테스트 누락 여부

출력 형식:
{{
  "passed": true/false,
  "critical_issues": [],
  "warnings": [],
  "suggestions": [],
  "summary": "전체 요약 (마크다운)"
}}
""")

    # 결과 파싱
    import re
    match = re.search(r'\{.*\}', response.text, re.DOTALL)
    if match:
        result = json.loads(match.group())
    else:
        result = {"passed": True, "summary": response.text}

    # CI용 결과 파일 저장
    with open("review_result.json", "w", encoding="utf-8") as f:
        json.dump(result, f, ensure_ascii=False)

    print(json.dumps(result, ensure_ascii=False, indent=2))

    # Critical 이슈 있으면 CI 실패
    if not result.get("passed") and result.get("critical_issues"):
        print("❌ Critical 이슈 발견 — PR 차단")
        exit(1)

asyncio.run(run_pr_review())

# .github/workflows/ai-agent-deploy.yml — 배포 에이전트
name: AI Agent Deploy

on:
  push:
    branches: [main]

jobs:
  agent-deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Antigravity CLI 설치
        run: npm install -g @google/antigravity-cli

      - name: AI 에이전트로 배포 실행
        env:
          GEMINI_API_KEY: ${{ secrets.GEMINI_API_KEY }}
          GCP_PROJECT: ${{ secrets.GCP_PROJECT }}
        run: |
          antigravity "
            main 브랜치 최신 코드로:
            1. Docker 이미지 빌드 (태그: $(git rev-parse --short HEAD))
            2. Artifact Registry에 푸시
            3. Cloud Run에 카나리 배포 (트래픽 10%)
            4. 헬스체크 5분 동안 모니터링
            5. 정상이면 100%로 전환, 비정상이면 롤백
            6. 완료 결과를 #deployments Slack 채널에 알림
          " --sandbox --timeout 600

[CI/CD 에이전트 패턴 정리]

PR 에이전트 (코드 리뷰):
→ on: pull_request 트리거
→ diff 내용 → 에이전트 분석 → PR 코멘트 자동 게시
→ Critical 이슈 → exit(1) → 머지 차단

배포 에이전트:
→ on: push to main 트리거
→ 빌드 → 푸시 → 카나리 배포 → 검증 → 전환
→ 실패 시 자동 롤백 + Slack 알림

보안 스캔 에이전트:
→ on: schedule (야간)
→ 전체 코드베이스 취약점 스캔
→ 결과를 Security 이슈로 자동 생성

---

실전6 — Hooks 시스템: 에이전트 실행 전후 제어

# hooks_example.py — SDK Hooks로 에이전트 행동 제어
from google.antigravity import Agent, LocalAgentConfig
from google.antigravity.hooks import HookRunner, before_tool, after_tool, on_error
import asyncio, logging

logger = logging.getLogger(__name__)

# ── 툴 실행 전 훅 ────────────────────────────────────────
@before_tool("shell")
async def validate_shell_command(command: str, **kwargs):
    """위험한 셸 명령 실행 전 차단"""
    DANGEROUS_CMDS = ["rm -rf", "drop table", "delete from", "format c:"]
    if any(cmd in command.lower() for cmd in DANGEROUS_CMDS):
        raise PermissionError(f"위험한 명령 차단됨: {command}")
    logger.info(f"[Shell 실행 허용] {command[:50]}...")

# ── 툴 실행 후 훅 ────────────────────────────────────────
@after_tool("file_management")
async def log_file_changes(result: dict, **kwargs):
    """파일 변경사항 감사 로그"""
    if result.get("action") == "write":
        logger.info(f"[파일 쓰기] {result.get('path')} ({result.get('size')} bytes)")
        # 감사 DB에 기록
        await record_audit_log(
            action="file_write",
            path=result.get("path"),
            agent_session=result.get("session_id"),
        )

# ── 에러 훅 ──────────────────────────────────────────────
@on_error
async def handle_agent_error(error: Exception, context: dict):
    """에러 발생 시 Slack 알림"""
    await post_slack_message(
        "#agent-errors",
        f"❌ 에이전트 에러: {str(error)}\n컨텍스트: {context.get('task', 'Unknown')}"
    )

# ── 훅 적용해서 에이전트 실행 ────────────────────────────
async def run_with_hooks():
    config = LocalAgentConfig(
        model="gemini-3.5-flash",
        hooks=HookRunner(hooks=[
            validate_shell_command,
            log_file_changes,
            handle_agent_error,
        ]),
        credential_masking=True,
    )

    async with Agent(config=config) as agent:
        response = await agent.chat(
            "src/ 디렉토리 정리하고 불필요한 파일 삭제해줘"
        )
        print(response.text)

asyncio.run(run_with_hooks())

→ before_tool: 에이전트가 툴 호출 전 검증 — 위험 명령 차단
→ after_tool: 툴 실행 후 감사 로그 — 컴플라이언스 필수
→ on_error: 에러 처리 + 알림 — 프로덕션 필수
→ Hooks가 없으면 에이전트가 뭘 하는지 통제 불가
→ 엔터프라이즈: before_tool에서 승인 워크플로우 연동 가능
   (Jira 티켓 없으면 프로덕션 파일 수정 차단 등)

---

✅ 핵심 정리
✅ Managed Agents API: 목표만 선언 → Google이 샌드박스·루프·재시도 전부 관리
✅ 과금: 실행 단위 — 장기 에이전트 작업에 토큰 단위보다 유리
✅ MCP 서버 연동: GitHub·BigQuery·Cloud Run 직접 조작 가능
✅ GCP Enterprise: VPC 프라이빗 실행 + IAM + 감사 로그 — 규정 준수 환경 필수
✅ GitHub Actions 통합: PR마다 에이전트 자동 리뷰, Critical 이슈 머지 차단
✅ Hooks 시스템: before_tool로 위험 명령 차단, after_tool로 감사 로그

❌ API 키를 에이전트 컨텍스트에 직접 전달 금지 — credential_masking: True 필수
❌ GitHub 레포 클론만으로 SDK 설치 안 됨 — 반드시 PyPI에서 pip install
❌ Free Tier: SLA·VPC 없음 — 프로덕션은 GCP Enterprise Platform 사용
❌ Hooks 없이 파일 삭제 작업 맡기면 통제 불가 — before_tool 검증 필수

---

관련글

• Google Antigravity 2.0 완전 정리 — Gemini CLI 마이그레이션 가이드

 

 

Google I/O 2026 핵심 정리 2편 — Antigravity 2.0, 개발자 도구 총정리, Android XR 안경 하드웨어