深入浅出Managed agent,从本地隔离到云端托管 (Google Antigravity cli 实战指南)

+-----------------------------------------------------------------------------------+
|                            宿主与沙盒物理执行边界                                 |
|                                                                                   |
|  +-----------------------------------------------------------------------------+  |
|  |                Google Cloud / Local Linux Sandbox (执行环境)                |  |
|  |                                                                             |  |
|  |    +-------------------------------------------------------------------+    |  |
|  |    | 1. 密钥配置层 (Configuration)                                     |    |  |
|  |    |    - 映射 .env 文件 (TELEGRAM_BOT_TOKEN / CHAT_ID)                |    |  |
|  |    +------------------+------------------------------------------------+    |  |
|  |                       | (读取注入)                                          |  |
|  |                       v                                                     |  |
|  |    +-------------------------------------------------------------------+    |  |
|  |    | 2. 推理与控制核心 (Execution Core)                                  |    |  |
|  |    |    - 一次性执行引擎 (python3 telegram_cron_simulator.py --once)     |    |  |
|  |    +------------------+------------------------------------------------+    |  |
|  |                       | (自检触发)                                          |  |
|  |                       v                                                     |  |
|  |    +-------------------------------------------------------------------+    |  |
|  |    | 3. 多维自检与遥测引擎 (Telemetry Engine)                           |    |  |
|  |    |    - 纯 urllib 接口                                               |    |  |
|  |    |    - ipinfo.io API (获取出口IP及Toronto等位置)                      |    |  |
|  |    |    - wttr.in 气象 API (抓取温度及晴雨)                              |    |  |
|  |    +------------------+------------------------------------------------+    |  |
|  |                       | (TLS 1.3 安全发送)                                  |  |
|  |                       v                                                     |  |
|  |    +-------------------------------------------------------------------+    |  |
|  |    | 4. 出口安全网络层 (Egress Network)                                 |    |  |
|  |    |    - 沙盒网格 VPC 虚拟网卡 (网络白名单限制)                             |    |  |
|  |    +------------------+------------------------------------------------+    |  |
|  +-----------------------|-----------------------------------------------------+  |
+--------------------------|--------------------------------------------------------+
                           |
                           | (SSL / TLS 1.3 链路安全穿越)
                           v
        +------------------+------------------+
        |  电报云端服务器 (Telegram Bot API)  |
        +------------------+------------------+
                           |
                           | (Push 消息推送)
                           v
        +------------------+------------------+
        |   用户物理终端 (Telegram Mobile App) |
        +-------------------------------------+

TELEGRAM_BOT_TOKEN=8763417888:AAE3U-VFaM4FYB5dpEb6xlRBiHdMx16v888
TELEGRAM_CHAT_ID=8250511888

#!/usr/bin/env python3
import os
import sys
import time
import urllib.request
import urllib.parse
import json
from datetime import datetime

def load_env():
    env_file = ".env"
    if not os.path.exists(env_file):
        env_file = os.path.join(os.path.dirname(__file__), ".env")
        if not os.path.exists(env_file):
            return {}
    config = {}
    with open(env_file, "r") as f:
        for line in f:
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            if "=" in line:
                key, val = line.split("=", 1)
                config[key.strip()] = val.strip()
    return config

def get_sandbox_network_info():
    """通过 ipinfo.io 探测沙盒的公网 IP 和地理位置"""
    try:
        req = urllib.request.Request("https://ipinfo.io/json", headers={'User-Agent': 'Mozilla/5.0'})
        with urllib.request.urlopen(req, timeout=5) as response:
            data = json.loads(response.read().decode())
            return {
                "ip": data.get("ip", "Unknown IP"),
                "city": data.get("city", "Unknown City"),
                "region": data.get("region", "Unknown Region"),
                "country": data.get("country", "Unknown Country"),
                "org": data.get("org", "Unknown ISP")
            }
    except Exception as e:
        return {"ip": "Unknown IP", "city": "Unknown City", "region": "Unknown Region", "country": "Unknown Country", "org": str(e)}

def get_weather(city):
    """利用 wttr.in 免 Key 探测当地实时天气"""
    if not city or city == "Unknown City":
        url = "https://wttr.in?format=%C+%t"
    else:
        city_encoded = urllib.parse.quote(city)
        url = f"https://wttr.in/{city_encoded}?format=%C+%t"
    try:
        req = urllib.request.Request(url, headers={'User-Agent': 'curl/7.79.1'})
        with urllib.request.urlopen(req, timeout=5) as response:
            return response.read().decode('utf-8').strip()
    except Exception as e:
        return f"Weather unavailable ({str(e)})"

def send_telegram_message(token, chat_id, message):
    url = f"https://api.telegram.org/bot{token}/sendMessage"
    payload = urllib.parse.urlencode({"chat_id": chat_id, "text": message, "parse_mode": "Markdown"}).encode("utf-8")
    req = urllib.request.Request(url, data=payload, headers={"User-Agent": "Mozilla/5.0"})
    try:
        with urllib.request.urlopen(req, timeout=10) as response:
            res_data = json.loads(response.read().decode())
            return res_data.get("ok"), "Success"
    except Exception as e:
        return False, str(e)

def main():
    config = load_env()
    token = config.get("TELEGRAM_BOT_TOKEN")
    chat_id = config.get("TELEGRAM_CHAT_ID")
    
    if not token or not chat_id:
        print("[!] Error: Credentials missing in .env")
        sys.exit(1)
        
    args = sys.argv[1:]
    if args and args[0] == "--once":
        current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        net_info = get_sandbox_network_info()
        weather = get_weather(net_info["city"])
        
        message = (
            f"🔔 *Antigravity Cloud Sandbox Explorer Update*\n\n"
            f"📍 *位置 (Location)*: `{net_info['city']}, {net_info['region']}, {net_info['country']}`\n"
            f"🌐 *公网 IP (Exit IP)*: `{net_info['ip']}`\n"
            f"🏢 *运营商 (ISP)*: `{net_info['org']}`\n"
            f"🌤️ *当地天气 (Weather)*: `{weather}`\n"
            f"⏰ *当前时间 (Time)*: `{current_time}`\n\n"
            f"🚀 *状态*: 沙盒多维探测运行正常！"
        )
        
        success, info = send_telegram_message(token, chat_id, message)
        if success:
            print("  --> SUCCESS! Probed metadata delivered to your Telegram Bot.")
        else:
            print(f"  --> FAILED: {info}")

if __name__ == "__main__":
    main()

from google import genai
import os

# 1. 初始化官方 SDK 客户端 (自动从 GEMINI_API_KEY 环境变量加载)
client = genai.Client()

agent_id = "telegram-notifier"

# 2. 🧹 清理可能存在的旧配置，保证部署的幂等性
try:
    client.agents.delete(id=agent_id)
except Exception:
    pass

# 3. 🚀 部署并注册 Custom Agent 蓝图 (包含系统提示词、源码文件和技能)
client.agents.create(
    id=agent_id,
    base_agent="antigravity-preview-05-2026",
    system_instruction="你是一个云端一次性遥测智能体...",
    base_environment={
        "type": "remote",
        "sources": [
            {
                "type": "inline",
                "target": ".env",
                "content": "TELEGRAM_BOT_TOKEN=...\nTELEGRAM_CHAT_ID=...\n"
            },
            {
                "type": "inline",
                "target": "telegram_cron_simulator.py",
                "content": "[Python 零依赖探针源码]"
            },
            {
                "type": "inline",
                "target": ".agents/skills/telegram-pusher/SKILL.md",
                "content": "[技能描述，定义了运行 python3 telegram_cron_simulator.py --once]"
            }
        ]
    }
)

# 4. 🔌 一键拉起一次性无服务器云端沙盒运行，发送消息后沙盒物理销毁
interaction = client.interactions.create(
    agent=agent_id,
    input="Execute the telegram-pusher skill to collect remote telemetry.",
    environment="remote" # 👈 直接传 "remote" 字符串，云端会自动根据预注册的 blueprint 拉起沙盒！
)

print("云端执行完成！请检查您的 Telegram 手机端。")

{
  "antigravity.sandbox.enabled": true,
  "antigravity.sandbox.allowedDirectories": [
    "/Users/eric/work/agy"
  ]
}

{
  "permissions": {
    "command": {
      "allowed": [
        "echo", "date", "ls" // 🟢 允许自动静默执行，绝不弹窗
      ],
      "ask": [
        "brew", "pip", "python3.11", "curl" // 🟡 触发黄色气泡，必须人类手动按 1 授权
      ],
      "denied": [
        "sudo", "rm -rf /", "chmod" // 🔴 物理死锁，直接报错，不允许运行
      ]
    },
    "read_file": {
      "allowed": ["/Users/eric/work/agy"],
      "denied": ["/Users/eric/.ssh", "/Users/eric/.gemini/config/config.json"]
    }
  }
}