飞书远程控制 Antigravity (AGY) 操作与设计文档

本文档详细介绍了如何通过飞书机器人远程操控本地 Google Antigravity (agy) 智能代码助手的原理、架构设计、配置部署以及使用方法。

---

一、 系统工作原理与架构

1.1 总体架构图

+------------------+         WebSocket 长连接          +-------------------------+
|  飞书客户端      | <==============================> |  feishu_agt.py 桥接服务  |
| (私聊 / 群聊 @)  |      (接收事件 & 回传消息)        |  (Lark OAPI WebSocket)  |
+------------------+                                  +-------------------------+
                                                                   |
                                                      子线程 / 子进程 (subprocess)
                                                                   |
                                                                   v
                                                      +-------------------------+
                                                      |  Antigravity CLI (agy)  |
                                                      |  --dangerously-skip-    |
                                                      |  permissions -p |
                                                      +-------------------------+
                                                                   |
                                                      读取/编写代码 & 执行工具
                                                                   |
                                                                   v
                                                      +-------------------------+
                                                      |  本地项目工作区目录     |
                                                      |  (如 E:\ai\project\...) |
                                                      +-------------------------+

1.2 核心原理解析

- 采用 lark.ws.Client，无需公网 IP 和 Webhook 域名，脚本启动后主动与飞书服务器建立长连接，实时接收用户消息。
- 自动支持飞书单聊与群聊（群聊中通过 @机器人 触发）。

1. 飞书长连接（WebSocket 模式）：

- 支持通过 /cd <路径> 为每个飞书会话（chat_id）独立配置工作目录，不同群或不同人的任务互不干扰。
- 消息通过 message_id 去重缓存，防止网络重连时事件重复消费。

1. 指令路由与会话状态隔离：

- 借助 agy -p "" 以非交互模式运行任务。
- 必须携带 --dangerously-skip-permissions 参数，自动批准 Agent 的文件编辑、命令运行等操作，避免后台进程因等待终端确认而挂死。
- 设定 10 分钟任务超时控制和异常捕获，确保服务长期稳定运行。

1. Antigravity CLI (agy) 非交互调用：

- 针对飞书单条消息约 4000 字符的限制，执行结果若过长将自动切片分批发送，防止关键输出截断。

1. 消息分片与格式优化：

---

二、 飞书开放平台配置

在使用本脚本前，需确保在飞书开发者后台正确创建并配置了应用：

- 访问 飞书开放平台 -> 进入开发者后台 -> 创建企业自建应用。
- 获取应用的 App ID 和 App Secret。

1. 创建企业自建应用：

- 在应用详情页左侧导航进入「添加应用能力」 -> 开启「机器人」。

1. 配置机器人能力：

- im:message（获取与发送单聊/群聊消息）
- im:message.p2p_msg:readonly（读取用户发给机器人的单聊消息）
- im:message.group_at_msg:readonly（读取群聊中 @ 机器人的消息）
- im:message:send_as_bot（以应用身份发送消息）

1. 开通必要权限（权限管理）：

- 进入「事件与回调」页面：
- 订阅方式选择 「使用 WebSocket 长连接」。
- 添加事件：「接收消息 v2.0」 (im.message.receive_v1)。
- 发布应用版本生效。

1. 开启事件订阅（长连接模式）：

---

三、 环境依赖与准备

1. Python 依赖安装

pip install lark-oapi

feishu_agt.py内容

import json
import os
import re
import subprocess
import threading
from collections import OrderedDict
import lark_oapi as lark
from lark_oapi.api.im.v1 import (
    P2ImMessageReceiveV1,
    CreateMessageRequest,
    CreateMessageRequestBody,
)

# 支持从环境变量读取，或使用默认配置
APP_ID = os.getenv("FEISHU_APP_ID", "cli_xxx")
APP_SECRET = os.getenv("FEISHU_APP_SECRET", "xxx")

# 默认初始工作区
DEFAULT_WORKSPACE = os.path.expanduser("E:\\ai\\project")

# 记录不同飞书会话的工作目录 {chat_id: current_path}
session_workspaces = {}

# 消息去重缓存（避免网络重连时重复消费事件）
processed_messages = OrderedDict()
MAX_CACHE_SIZE = 1000

client = lark.Client.builder().app_id(APP_ID).app_secret(APP_SECRET).build()


def send_reply(chat_id: str, text: str):
    """回传消息到飞书，自动处理长文本切片发送"""
    chunk_size = 3500
    chunks = [text[i : i + chunk_size] for i in range(0, len(text), chunk_size)] or [text]

    for chunk in chunks:
        req = (
            CreateMessageRequest.builder()
            .receive_id_type("chat_id")
            .request_body(
                CreateMessageRequestBody.builder()
                .receive_id(chat_id)
                .msg_type("text")
                .content(json.dumps({"text": chunk}))
                .build()
            )
            .build()
        )
        try:
            client.im.v1.message.create(req)
        except Exception as e:
            print(f"[-] 消息发送失败: {e}")


def get_current_workspace(chat_id: str) -> str:
    """获取当前会话的工作目录"""
    return session_workspaces.get(chat_id, DEFAULT_WORKSPACE)


def run_antigravity_task(prompt: str, chat_id: str, workspace: str):
    """在指定目录下调用 Antigravity (agy) 执行任务"""
    send_reply(
        chat_id,
        f"📂 工作目录: `{workspace}`\n🚀 收到任务，开始调用 Antigravity (agy) 执行：\n> {prompt}",
    )

    try:
        # 正确调用 agy CLI：
        # -p: 以 print 模式非交互执行 prompt
        # --dangerously-skip-permissions: 自动批准工具调用与写文件，避免因等待交互导致挂起
        cmd = ["agy", "-p", prompt, "--dangerously-skip-permissions"]

        process = subprocess.Popen(
            cmd,
            cwd=workspace,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            encoding="utf-8",
            errors="replace",
            shell=True,
        )

        # 设置 10 分钟任务超时
        stdout, _ = process.communicate(timeout=600)
        status = "✅ 执行完毕" if process.returncode == 0 else "❌ 执行异常/报错"

        result_text = stdout.strip() if stdout else "（无控制台输出）"
        send_reply(
            chat_id,
            f"{status} (目录: `{os.path.basename(workspace)}`)\n\n{result_text}",
        )

    except subprocess.TimeoutExpired:
        process.kill()
        send_reply(chat_id, "⚠️ 执行超时（超过 10 分钟），任务已被强制终止。")
    except Exception as e:
        send_reply(chat_id, f"⚠️ 执行异常: {str(e)}")


def clean_feishu_prompt(raw_text: str) -> str:
    """清理飞书 @ 机器人的占位符及多余前缀"""
    cleaned = re.sub(r"^@_user_\w+\s*", "", raw_text).strip()
    if cleaned.startswith("@"):
        cleaned = re.sub(r"^@\S+\s*", "", cleaned).strip()
    return cleaned


def handle_command(text: str, chat_id: str):
    """处理用户指令或普通提示词"""
    parts = text.split(maxsplit=1)
    cmd = parts[0].lower()
    arg = parts[1].strip() if len(parts) > 1 else ""

    # 1. 切换工作目录
    if cmd in ["/cd", "/setdir", "/workspace"]:
        if not arg:
            send_reply(
                chat_id,
                "⚠️ 请提供目标路径，例如：\n`/cd E:\\ai\\project\\feishu_agt`\n`/cd D:\\workspace\\my-project`",
            )
            return

        target_path = os.path.abspath(os.path.expanduser(arg))
        if os.path.isdir(target_path):
            session_workspaces[chat_id] = target_path
            send_reply(
                chat_id,
                f"✅ 工作目录已成功切换为：\n`{target_path}`\n后续发送的任务将在此目录下执行。",
            )
        else:
            send_reply(
                chat_id,
                f"❌ 切换失败，目录不存在或无权访问：\n`{target_path}`",
            )
        return

    # 2. 查询当前工作目录
    if cmd in ["/pwd", "/dir"]:
        curr = get_current_workspace(chat_id)
        send_reply(chat_id, f"📂 当前会话工作目录为：\n`{curr}`")
        return

    # 3. 帮助菜单
    if cmd in ["/help", "/h", "帮助"]:
        help_msg = (
            "📌 **Antigravity 飞书控制指令说明**：\n"
            "• `/cd <路径>`：切换当前会话的工作目录（如 `/cd E:\\ai\\project`）\n"
            "• `/pwd`：查看当前所在的工作目录\n"
            "• `/help`：查看指令说明\n"
            "• 直接发送其他任何文本：将在当前目录下交由 Antigravity (agy) 自动执行"
        )
        send_reply(chat_id, help_msg)
        return

    # 4. 普通自然语言 Prompt -> 启动线程异步执行
    current_dir = get_current_workspace(chat_id)
    threading.Thread(
        target=run_antigravity_task,
        args=(text, chat_id, current_dir),
        daemon=True,
    ).start()


def message_handler(data: P2ImMessageReceiveV1) -> None:
    """处理飞书接收到的消息事件"""
    message = data.event.message
    msg_id = message.message_id

    # 消息去重
    if msg_id in processed_messages:
        return
    processed_messages[msg_id] = True
    if len(processed_messages) > MAX_CACHE_SIZE:
        processed_messages.popitem(last=False)

    if message.message_type == "text":
        content_dict = json.loads(message.content)
        raw_text = content_dict.get("text", "").strip()

        prompt = clean_feishu_prompt(raw_text)
        if prompt:
            handle_command(prompt, message.chat_id)


event_handler = (
    lark.EventDispatcherHandler.builder("", "")
    .register_p2_im_message_receive_v1(message_handler)
    .build()
)

if __name__ == "__main__":
    print("🤖 飞书 -> Antigravity (agy) 动态目录控制桥接已启动...")
    ws_client = lark.ws.Client(
        app_id=APP_ID,
        app_secret=APP_SECRET,
        event_handler=event_handler,
        log_level=lark.LogLevel.INFO,
    )
    ws_client.start()

2. Antigravity CLI 验证与首次登录认证

确保本地环境变量中已存在 agy 命令：

agy --help

*如未在 PATH 中，可检查安装目录（如 C:\Users\<用户名>\AppData\Local\agy\bin\）。*

[!IMPORTANT]

首次运行必须在本地终端完成一次性登录：

1. 在终端（PowerShell 或 CMD）中运行 agy。

2. 浏览器会自动打开 Google 授权登录页面，登录后复制授权码（如 4/0ATs...）。

3. 回到终端粘贴授权码并回车完成登录，凭据将自动持久化保存在本地。

4. 请勿在飞书中直接发送授权码，飞书机器人会将授权码作为普通任务下发，无法完成交互式认证。

---

四、 快速启动与网络代理配置

因 Antigravity 需访问 Google 云端服务（oauth2.googleapis.com 等），在国内网络环境下需确保终端网络畅通。

方式一：直接运行（读取脚本默认配置，推荐配合代理软件 TUN 模式）

python feishu_agt.py

方式二：在终端注入代理环境变量后运行

若未开启 TUN 虚拟网卡模式，需在 PowerShell 终端先设置代理端口（以常见的 7890 端口为例）：

# 1. 设置终端代理环境变量
$env:HTTP_PROXY="http://127.0.0.1:7890"
$env:HTTPS_PROXY="http://127.0.0.1:7890"

# 2. （可选）自定义飞书凭据
# $env:FEISHU_APP_ID="cli_xxxxxxxxxxxx"
# $env:FEISHU_APP_SECRET="yyyyyyyyyyyyyyyy"

# 3. 启动桥接服务
python feishu_agt.py

控制台看到如下输出即表示启动成功：

🤖 飞书 -> Antigravity (agy) 动态目录控制桥接已启动...

---

五、 飞书端操作方式与命令手册

在飞书找到机器人（或拉入群聊），即可通过以下方式进行交互：

| 指令 / 消息 | 说明 | 示例 |
| :--- | :--- | :--- |
| /help 或 帮助 | 查看指令帮助说明 | /help |
| /pwd 或 /dir | 查看当前会话绑定的工作目录 | /pwd |
| /cd <路径> | 切换当前会话的项目路径 | /cd E:\ai\project\my-web |
| 任意自然语言 | 直接向 Antigravity 派发开发任务 | 分析当前目录结构并写一个测试用例 |

操作流程演示：

> 用户：/cd E:\ai\project\feishu_agt
> 机器人：✅ 工作目录已成功切换为：E:\ai\project\feishu_agt

1. 步骤 1：查看或切换目录

> 用户：请查看当前目录下所有 py 文件并做语法检查
> 机器人：
> 📂 工作目录: E:\ai\project\feishu_agt
> 🚀 收到任务，开始调用 Antigravity (agy) 执行：
> > 请查看当前目录下所有 py 文件并做语法检查
>
> *(稍等片刻后输出结果)*
> ✅ 执行完毕 (目录: feishu_agt)
> 已检查 feishu_agt.py，语法正常无错误。

1. 步骤 2：下发任务

---

六、 常见问题排查 (FAQ)

1. 登录报错 token exchange failed: Post "https://oauth2.googleapis.com/token": dial tcp ...？

- 方案 A（最推荐）：在代理客户端（如 Clash、v2rayN、Sing-box 等）中开启 「TUN 模式」 或 「虚拟网卡增强模式」，全局接管终端网络。
- 方案 B：在当前 PowerShell 窗口中先执行 $env:HTTP_PROXY="http://127.0.0.1:7890"; $env:HTTPS_PROXY="http://127.0.0.1:7890"（端口根据实际情况修改），再执行 agy 登录。

• 原因：Windows 命令行终端默认不走系统代理，导致与 Google 授权接口连接超时。
• 解决办法：

2. 飞书报错 Authentication required. Please visit the URL to log in... 并超时？

• 原因：本地尚未完成首次 CLI 登录。feishu_agt.py 是非交互后台进程，无法实时读取你在飞书中粘贴的授权码。
• 解决办法：在本地 PowerShell 中手动运行一次 agy，在终端里粘贴授权码完成登录即可，以后无需重复登录。

3. 任务一直显示“开始执行”但没有结果？

• 检查终端网络是否能正常访问 Google 服务。
• 检查任务是否因缺少 --dangerously-skip-permissions 而在控制台等待交互确认（当前脚本已默认内置）。

4. 飞书提示“切换失败，目录不存在”？

• 请提供完整的 Windows 绝对路径，如 E:\ai\project。

5. 飞书群聊中发送消息机器人无反应？

• 确保在群聊中有 @机器人。
• 确认飞书开发者后台的「接收消息」事件订阅已发布最新版本。