MailUI4Agents/deploy/remote-agent-demo.py

#!/usr/bin/env python3
"""最小跨主机 Agent —— 只用 python 标准库，证明跨主机不需要新组件。

用途：在一台**没有装 AgentMail 任何代码**的机器上收发邮件。
如果这个脚本能跑通，「跨主机 Agent」在协议层面就已经成立 ——
不需要注册中心，因为连接方向是单向的：Agent 主动连 Gateway，Gateway 从不外呼。

用法：

    export AGENTMAIL_GATEWAY_URL=https://mail.example.com/api/v1
    export AGENTMAIL_AGENT_KEY=<在 Gateway 上建的 Agent 密钥>
    export AGENTMAIL_AGENT_NAME=remotebot
    export AGENTMAIL_WORKSPACE=/tmp/remotebot-ws     # 可选
    python3 remote-agent-demo.py [运行秒数]

它做四件事：注册 → 心跳（带模型目录）→ SSE 长连 → 收到邮件就回一封。
真正的插件还要做权限转发、会话命名回写、附件等，见 docs/PLUGIN-CONTRACT.md。
"""
import json
import os
import sys
import threading
import time
import urllib.error
import urllib.request

GW = os.environ.get("AGENTMAIL_GATEWAY_URL", "http://127.0.0.1:8180/api/v1").rstrip("/")
KEY = os.environ.get("AGENTMAIL_AGENT_KEY", "")
NAME = os.environ.get("AGENTMAIL_AGENT_NAME", "remotebot")
WORKSPACE = os.environ.get("AGENTMAIL_WORKSPACE", "/tmp/%s-ws" % NAME)

if not KEY:
    sys.exit("需要 AGENTMAIL_AGENT_KEY（在 Gateway 的管理页或 admin API 上建）")


def api(path, body=None, method=None):
    """打一次 Gateway。密钥走 Authorization 头，与插件完全一致。"""
    req = urllib.request.Request(
        GW + path,
        data=json.dumps(body).encode() if body is not None else None,
        headers={"Authorization": "Bearer " + KEY, "Content-Type": "application/json"},
        method=method or ("POST" if body is not None else "GET"),
    )
    try:
        with urllib.request.urlopen(req, timeout=20) as r:
            return json.loads(r.read() or b"{}")
    except urllib.error.HTTPError as e:
        # Gateway 的错误体是 {"error": "..."}，把它读出来 ——
        # 不读的话只剩一个 "HTTP Error 400"，看不出是哪个字段不对
        detail = e.read().decode(errors="replace")[:300]
        raise SystemExit("%s %s -> HTTP %s %s" % (req.method, path, e.code, detail))


def register():
    """注册。两个容易踩的地方：

    - 路由是 `/agent/register`（**单数**），复数会 404
    - `workspaces` 是对象数组 `[{name, path}]`，不是字符串数组；
      传字符串会得到 400 "请求体 JSON 解析失败"
    """
    return api("/agent/register", {
        "name": NAME,
        "platform": "stdlib-demo",
        "workspaces": [{"name": "demo", "path": WORKSPACE}],
    })


def heartbeat():
    """心跳。models 是这台机器上「有」的模型目录；
    响应回传管理员划定的范围，真正的插件据此决定按什么顺序尝试。"""
    return api("/agent/heartbeat", {
        "models": [
            {"provider": "stdlib", "model": "echo-1", "display_name": "Echo（演示）"},
        ],
    })


def reply_to_unread(event):
    """收到 new_mail 就把未读的都回一封，然后标记已读。

    标记已读不能省：不标的话下一轮会把同一批邮件再捞一次。
    """
    box = api("/mail/inbox?status=unread&limit=5")
    mails = box.get("mails", box if isinstance(box, list) else [])
    for m in mails:
        # B-5.6：收件方是 Agent → 不自动转发。from_human 由服务端判定，
        # 不依赖插件自己的猜测（早期靠 "from_name == 'human'" 的写法
        # 在多用户下恒为假，导致回信发给了自己）。
        if not m.get("from_human", True):
            print("  跳过 Agent 来信:", m.get("from_name"), m.get("subject"))
            api("/mail/read", {"mail_ids": [m.get("mail_id")]})
            continue
        print("  收到:", m.get("subject"), "| 工作目录:", event.get("to_workspace"))
        api("/mail/send", {
            "to": m.get("from_name"),
            "subject": "Re: " + (m.get("subject") or ""),
            "body": (
                "这封回信来自另一台主机上的一个纯标准库脚本，"
                "它没有安装 AgentMail 的任何代码。\n\n"
                "- 收到的工作目录: `%s`\n"
                "- 原邮件 ID: `%s`\n\n"
                "跨主机在协议层面已经成立：Agent 主动连 Gateway，"
                "只需要一个公网 URL 加一把密钥。"
                % (event.get("to_workspace"), m.get("mail_id"))
            ),
            "reply_to": m.get("mail_id"),
        })
        api("/mail/read", {"mail_ids": [m.get("mail_id")]})
        print("  已回信")


def stream():
    """SSE 长连。真正的插件还要处理断线重连与 Last-Event-ID 补投。"""
    req = urllib.request.Request(
        GW + "/events/stream",
        headers={"Authorization": "Bearer " + KEY, "Accept": "text/event-stream"},
    )
    with urllib.request.urlopen(req, timeout=300) as r:
        etype = None
        for raw in r:
            line = raw.decode(errors="replace").rstrip("\n")
            if line.startswith("event: "):
                etype = line[7:]
            elif line.startswith("data: ") and etype:
                data = json.loads(line[6:])
                print("SSE <-", etype, json.dumps(data, ensure_ascii=False)[:100])
                if etype == "new_mail":
                    try:
                        reply_to_unread(data)
                    except Exception as e:      # noqa: BLE001 —— 一封失败不该断掉长连
                        print("  回信失败:", e)
                etype = None


def main():
    os.makedirs(WORKSPACE, exist_ok=True)
    print("Gateway:", GW, "| 身份:", NAME)
    print("注册:", json.dumps(register(), ensure_ascii=False)[:120])

    hb = heartbeat()
    print("心跳: pending=%s models_synced=%s allowed=%s unrestricted=%s" % (
        hb.get("pending_mails"),
        hb.get("models_synced"),
        [f"{m['provider']}/{m['model']}" for m in hb.get("allowed_models", [])],
        hb.get("models_unrestricted"),
    ))

    threading.Thread(target=stream, daemon=True).start()

    # 启动时先补拉一次：SSE 只推连上之后的事件，
    # 离线期间到的邮件只能从心跳的 pending_mails 得知。
    # 不补这一次的后果：重启前发的邮件永远不会被处理。
    if hb.get("pending_mails"):
        print("启动补拉 %s 封未读" % hb["pending_mails"])
        try:
            reply_to_unread({"to_workspace": WORKSPACE})
        except Exception as e:                  # noqa: BLE001
            print("  补拉失败:", e)

    seconds = float(sys.argv[1]) if len(sys.argv) > 1 else 120
    deadline = time.time() + seconds
    # 30 秒一次心跳：Gateway 靠 last_seen 判在线
    while time.time() < deadline:
        time.sleep(min(30, max(1, deadline - time.time())))
        try:
            heartbeat()
        except Exception as e:                  # noqa: BLE001
            print("心跳失败:", e)
    print("结束")


if __name__ == "__main__":
    main()