From fc330879826f3f0f37c16d6e60002e12e8dfb028 Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Thu, 3 Sep 2026 12:57:43 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20homeagent-mail-bridge=20=E2=80=94?= =?UTF-8?q?=E2=80=94=20TrueAgent=20=E6=8E=A5=E5=85=A5=20AgentMail=EF=BC=88?= =?UTF-8?q?MVP=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 架构 TrueAgent 是单一常驻 agent 事件循环，没有 per-conversation session。 插件像 QQ 插件一样：所有邮件注入同一个事件循环，靠中断消息文本传递 context。 不做 platform_sessions 上报（无 sessions 可映射）。 ## 工具 - read_inbox: 收件箱查询（渲染发件人/主题/正文/抄送/附件） - send_mail: 三维地址发信（支持 reply_to） - output_send__homeagent 输出通道（agent 可主动发信） ## 自动回信 InjectInputSync（阻塞）等待 agent 处理完毕 → 自动把 FinalText 发回给发件人（reply_to 指向原邮件）。agent 不需要记得调 send_mail。 试过 StageAfterOutput hook，但 before_output 看不到 ToolCalls （read_inbox 调用在 turn 中间就被清掉了），导致自动回信不触发。 InjectInputSync 更可靠：TrueAgent 保证不重入。 ## 基础设施 - 心跳 25 秒 - SSE 长连 + 自动重连（3 秒退避） - 注册时自动注册密钥（AGENTMAIL_AGENT_KEY 环境变量） - systemd drop-in 注入 AgentMail 环境变量 ## 构建 plugindev build → dist/homeagent_linux_amd64.hmap 安装到 /home/newqqagent/plugins/homeagent-mail-bridge/ homeagent.service drop-in: /etc/systemd/system/homeagent.service.d/agentmail.conf ## 端到端验证 admin → homeagent@/tmp/homeagent.new（自动回信验证） → homeagent 读取邮件 + 自动回信「收到确认」→ 25 秒往返完成 --- .gitignore | 1 + plugins/homeagent-mail-bridge/go.mod | 7 + plugins/homeagent-mail-bridge/main.go | 11 + plugins/homeagent-mail-bridge/plg.json | 15 + plugins/homeagent-mail-bridge/plugin.go | 578 ++++++++++++++++++++++ plugins/homeagent-mail-bridge/plugin.json | 13 + 6 files changed, 625 insertions(+) create mode 100644 plugins/homeagent-mail-bridge/go.mod create mode 100644 plugins/homeagent-mail-bridge/main.go create mode 100644 plugins/homeagent-mail-bridge/plg.json create mode 100644 plugins/homeagent-mail-bridge/plugin.go create mode 100644 plugins/homeagent-mail-bridge/plugin.json diff --git a/.gitignore b/.gitignore index 47b5b41..65512c2 100644 --- a/.gitignore +++ b/.gitignore @@ -58,3 +58,4 @@ attachments/ *.swp .idea/ .vscode/ +plugins/homeagent-mail-bridge/build/ diff --git a/plugins/homeagent-mail-bridge/go.mod b/plugins/homeagent-mail-bridge/go.mod new file mode 100644 index 0000000..2a71f33 --- /dev/null +++ b/plugins/homeagent-mail-bridge/go.mod @@ -0,0 +1,7 @@ +module homeagent-mail-bridge + +go 1.25.0 + +require gitcode.com/JianFeeeee/homeagent-sdk v0.0.0 + +replace gitcode.com/JianFeeeee/homeagent-sdk => /home/program/TrueAgent/third_party/homeagent-sdk diff --git a/plugins/homeagent-mail-bridge/main.go b/plugins/homeagent-mail-bridge/main.go new file mode 100644 index 0000000..67c3a3c --- /dev/null +++ b/plugins/homeagent-mail-bridge/main.go @@ -0,0 +1,11 @@ +//go:build !windows || !cgo + +package main + +import ( + sdk "gitcode.com/JianFeeeee/homeagent-sdk/sdk" +) + +func NewPlugin(name string, config map[string]interface{}) (sdk.Plugin, error) { + return NewPluginFactory(name, config) +} diff --git a/plugins/homeagent-mail-bridge/plg.json b/plugins/homeagent-mail-bridge/plg.json new file mode 100644 index 0000000..c6a0fcf --- /dev/null +++ b/plugins/homeagent-mail-bridge/plg.json @@ -0,0 +1,15 @@ +{ + "name": "homeagent-mail-bridge", + "name_zh": "AgentMail 桥接", + "name_en": "homeagent", + "version": "0.1.0", + "description": "HomeAgent 接入 AgentMail：邮件驱动的多智能体协作", + "author": "AgentMail", + "entry": "plugin.bin", + "tags": ["mail", "agentmail"], + "targets": "linux/amd64", + "outdir": "dist", + "bundle": false, + "replaces": {}, + "source_dirs": [] +} diff --git a/plugins/homeagent-mail-bridge/plugin.go b/plugins/homeagent-mail-bridge/plugin.go new file mode 100644 index 0000000..fa97d72 --- /dev/null +++ b/plugins/homeagent-mail-bridge/plugin.go @@ -0,0 +1,578 @@ +package main + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "log" + "net/http" + "os" + "strings" + "sync" + "time" + + "gitcode.com/JianFeeeee/homeagent-sdk/sdk" +) + +const ( + defaultGateway = "http://127.0.0.1:8180" + heartbeatIntval = 25 * time.Second + sseRetry = 3 * time.Second + mailPollIntval = 15 * time.Second +) + +// Plugin 实现 sdk.Plugin。 +// +// TrueAgent 是单一常驻 agent 事件循环，没有「每个对话一个 session」的概念。 +// 因此本插件不做会话映射 —— 所有邮件注入同一个事件循环，像 QQ 插件一样。 +// 邮件的 context 完全靠中断消息的文本传递，不靠 platform_sessions 上报。 +type Plugin struct { + name string + sdk *sdk.PluginSDK + gwURL string + key string + client *http.Client + stopCh chan struct{} + stopOnce sync.Once +} + +func NewPluginFactory(name string, config map[string]interface{}) (sdk.Plugin, error) { + gw := "" + key := "" + if v, ok := config["gateway_url"].(string); ok { + gw = v + } + if v, ok := config["gateway_key"].(string); ok { + key = v + } + if gw == "" { + gw = os.Getenv("AGENTMAIL_GATEWAY_URL") + } + if gw == "" { + gw = defaultGateway + } + if key == "" { + key = os.Getenv("AGENTMAIL_AGENT_KEY") + } + return &Plugin{ + name: name, + gwURL: strings.TrimRight(gw, "/"), + key: key, + client: &http.Client{Timeout: 30 * time.Second}, + stopCh: make(chan struct{}), + }, nil +} + +func (p *Plugin) Name() string { return p.name } + +func (p *Plugin) Start(s *sdk.PluginSDK) error { + p.sdk = s + s.SetAutoRestart(true) + + // 注册读取工具 + s.RegisterTool("read_inbox", sdk.ToolDef{ + Name: "read_inbox", + Description: "查阅收件箱中的邮件。收到新邮件通知后应立即调用此工具。每封含 mail_id、发件人、主题、正文与附件清单。", + Parameters: map[string]interface{}{ + "type": "object", + "properties": map[string]interface{}{ + "status": map[string]interface{}{ + "type": "string", + "description": "过滤条件 unread|all，默认 unread", + }, + "limit": map[string]interface{}{ + "type": "number", + "description": "返回数量，默认 5", + }, + }, + }, + }, p.handleReadInbox) + + // 注册发信工具 + s.RegisterTool("send_mail", sdk.ToolDef{ + Name: "send_mail", + Description: "发送邮件。三维地址 name@path.session。回复来信请传 reply_to。", + Parameters: map[string]interface{}{ + "type": "object", + "properties": map[string]interface{}{ + "to": map[string]interface{}{"type": "string", "description": "收件人三维地址"}, + "subject": map[string]interface{}{"type": "string", "description": "邮件主题"}, + "body": map[string]interface{}{"type": "string", "description": "邮件正文（Markdown）"}, + "cc": map[string]interface{}{"type": "string", "description": "抄送"}, + "reply_to": map[string]interface{}{"type": "string", "description": "回复某封邮件时传其 mail_id"}, + }, + "required": []string{"to", "subject", "body"}, + }, + }, p.handleSendMail) + + // 注册输出通道 —— agent 可以主动调 output_send__homeagent 发信 + s.RegisterOutputChannel("homeagent", sdk.CapText|sdk.CapFile, + "发送邮件。meta JSON 格式：{to, subject, reply_to}，type: text", + sdk.ChannelDef{}, p.handleOutputChannel) + + log.Printf("[homeagent-mail-bridge] 注册完成，等待 Gateway SSE") + + // 启动心跳 + SSE（后台 goroutine） + go p.heartbeatLoop() + go p.sseLoop() + + return nil +} + +func (p *Plugin) Stop() error { + p.stopOnce.Do(func() { close(p.stopCh) }) + return nil +} + +// ─── 心跳 ─── + +func (p *Plugin) heartbeatLoop() { + // 首次注册 + if err := p.register(); err != nil { + log.Printf("[homeagent-mail-bridge] 注册失败: %v", err) + } + ticker := time.NewTicker(heartbeatIntval) + defer ticker.Stop() + for { + select { + case <-p.stopCh: + return + case <-ticker.C: + if err := p.heartbeat(); err != nil { + log.Printf("[homeagent-mail-bridge] 心跳失败: %v", err) + } + } + } +} + +func (p *Plugin) register() error { + body := map[string]interface{}{ + "name": p.name, + "platform": "homeagent", + "workspaces": []interface{}{}, + } + return p.post("/agent/register", body, nil) +} + +func (p *Plugin) heartbeat() error { + return p.post("/agent/heartbeat", map[string]interface{}{}, nil) +} + +// ─── SSE ─── + +func (p *Plugin) sseLoop() { + for { + select { + case <-p.stopCh: + return + default: + } + if err := p.readSSE(); err != nil { + log.Printf("[homeagent-mail-bridge] SSE 断开: %v，%v 后重连", err, sseRetry) + time.Sleep(sseRetry) + } + } +} + +func (p *Plugin) readSSE() error { + req, err := http.NewRequest("GET", p.gwURL+"/api/v1/events/stream", nil) + if err != nil { + return err + } + req.Header.Set("Authorization", "Bearer "+p.key) + + resp, err := p.client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + if resp.StatusCode != 200 { + return fmt.Errorf("SSE HTTP %d", resp.StatusCode) + } + + log.Printf("[homeagent-mail-bridge] SSE 已连接") + + buf := make([]byte, 0, 4096) + lineStart := 0 + for { + select { + case <-p.stopCh: + return nil + default: + } + + n, err := resp.Body.Read(buf[len(buf):cap(buf)]) + if n > 0 { + buf = buf[:len(buf)+n] + // 处理完整行 + for { + i := bytes.IndexByte(buf[lineStart:], '\n') + if i < 0 { + break + } + line := string(buf[lineStart : lineStart+i]) + lineStart += i + 1 + p.parseSSELine(line) + } + if lineStart > 0 { + buf = buf[lineStart:] + lineStart = 0 + } + buf = buf[:len(buf)] + } + if err != nil { + if err != io.EOF { + return err + } + return nil + } + } +} + +func (p *Plugin) parseSSELine(line string) { + // SSE 格式：event: xxx\ndata: {...}\n\n + if !strings.HasPrefix(line, "data: ") { + return + } + raw := strings.TrimPrefix(line, "data: ") + if raw == "" || raw == "{}" { + return + } + + var evt struct { + MailID string `json:"mail_id"` + SessionID string `json:"session_id"` + FromName string `json:"from_name"` + Subject string `json:"subject"` + MailType string `json:"mail_type"` + Role string `json:"role"` + Workspace string `json:"to_workspace"` + Alias string `json:"session_alias"` + ReplyAddr string `json:"reply_address"` + } + if err := json.Unmarshal([]byte(raw), &evt); err != nil { + return + } + + if evt.MailID == "" { + return + } + + // 权限决策回复（MUST） + if evt.MailType == "permission_decision" { + p.handlePermissionDecision(evt) + return + } + + // 新邮件：注入 agent 事件循环 + if evt.MailType == "normal" { + p.handleNewMail(evt) + } +} + +// ─── 输出通道（agent 主动发信）─── + +func (p *Plugin) handleOutputChannel(args map[string]interface{}) (interface{}, error) { + payload, _ := args["payload"].(string) + meta, _ := args["meta"].(string) + if payload == "" { + return nil, fmt.Errorf("payload 不能为空") + } + + var m struct { + To string `json:"to"` + Subject string `json:"subject"` + ReplyTo string `json:"reply_to"` + } + if meta != "" { + json.Unmarshal([]byte(meta), &m) + } + if m.To == "" { + return nil, fmt.Errorf("meta 中需要 to 字段") + } + + if err := p.sendMail(m.To, m.Subject, payload, m.ReplyTo); err != nil { + return nil, err + } + return map[string]interface{}{"status": "sent"}, nil +} + +func (p *Plugin) sendMail(to, subject, body, replyTo string) error { + payload := map[string]interface{}{ + "to": to, + "subject": subject, + "body": body, + } + if replyTo != "" { + payload["reply_to"] = replyTo + } + return p.post("/mail/send", payload, nil) +} + +// ─── 新邮件处理 ─── + +func (p *Plugin) handleNewMail(evt struct { + MailID string `json:"mail_id"` + SessionID string `json:"session_id"` + FromName string `json:"from_name"` + Subject string `json:"subject"` + MailType string `json:"mail_type"` + Role string `json:"role"` + Workspace string `json:"to_workspace"` + Alias string `json:"session_alias"` + ReplyAddr string `json:"reply_address"` +}) { + // 记录回信目标 —— StageBeforeOutput 会检查是否由邮件触发， + // StageAfterOutput 用这个地址把最终回复发回去。 + // 构建中断消息 + prompt := fmt.Sprintf( + "你收到一封新邮件（AgentMail）。\n\n"+ + "发件人：%s\n"+ + "主题：%s\n"+ + "邮件 ID：%s\n"+ + "身份：你是 %s\n\n"+ + "请先调用 read_inbox 读取完整正文，然后处理其中的请求。\n\n"+ + "**回信不用你自己发**：你把本轮工作做完、把结论说出来就行，\n"+ + "插件会在这一轮结束时自动把你最后那段话作为回信发回给 %s（不消耗你的发信配额）。\n"+ + "只有在需要主动联系其他人、或要带附件时才调用 send_mail。", + evt.FromName, evt.Subject, evt.MailID, p.name, evt.FromName, + ) + + // 非阻塞注入 —— TrueAgent 的事件循环会处理 + // InjectInputSync 阻塞等待 agent 处理完毕，返回最终回复文本。 + // 这比 stage handler 更可靠：TrueAgent 的事件循环保证不会重入， + // 而 stage handler 的触发时机依赖 ToolCalls 的状态快照，实际测试中 + // before_output 看不到 read_inbox 调用（它在 turn 中间就被清掉了）。 + reply := p.sdk.InjectInputSync(p.name, p.name, prompt) + if reply == "" { + log.Printf("[homeagent-mail-bridge] 已注入新邮件 %s（来自 %s：%s），agent 无回复", evt.MailID[:8], evt.FromName, evt.Subject) + return + } + // 自动把 agent 的回复发回给发件人 + if err := p.sendMail(evt.FromName, "Re: "+evt.Subject, reply, evt.MailID); err != nil { + log.Printf("[homeagent-mail-bridge] 自动回信失败: %v", err) + } else { + log.Printf("[homeagent-mail-bridge] 已自动回信给 %s（%d 字）", evt.FromName, len(reply)) + } +} + +// ─── 权限决策 ─── + +func (p *Plugin) handlePermissionDecision(evt struct { + MailID string `json:"mail_id"` + SessionID string `json:"session_id"` + FromName string `json:"from_name"` + Subject string `json:"subject"` + MailType string `json:"mail_type"` + Role string `json:"role"` + Workspace string `json:"to_workspace"` + Alias string `json:"session_alias"` + ReplyAddr string `json:"reply_address"` +}) { + prompt := fmt.Sprintf( + "你之前发起的权限请求已有结论：%s（决策人：%s）。请据此继续。", + evt.Subject, evt.FromName, + ) + p.sdk.InjectText(p.name, p.name, prompt) +} + +// ─── 工具实现 ─── + +func (p *Plugin) handleReadInbox(args map[string]interface{}) (interface{}, error) { + status := "unread" + if v, ok := args["status"].(string); ok && v != "" { + status = v + } + limit := 5 + if v, ok := args["limit"].(float64); ok && v > 0 { + limit = int(v) + } + + url := fmt.Sprintf("%s/api/v1/mail/inbox?status=%s&limit=%d", p.gwURL, status, limit) + var result map[string]interface{} + if err := p.get(url, &result); err != nil { + return nil, err + } + + // 渲染成模型可读的文本 + mails, _ := result["mails"].([]interface{}) + if len(mails) == 0 { + return map[string]interface{}{"content": []map[string]interface{}{{"type": "text", "text": "收件箱为空。"}}}, nil + } + + var sb strings.Builder + for i, m := range mails { + mail, _ := m.(map[string]interface{}) + if mail == nil { + continue + } + from, _ := mail["from_name"].(string) + subj, _ := mail["subject"].(string) + mid, _ := mail["mail_id"].(string) + body, _ := mail["body"].(string) + if body == "" { + body, _ = mail["body_preview"].(string) + } + alias, _ := mail["session_alias"].(string) + + fmt.Fprintf(&sb, "[%d] %s: %s\n邮件 ID: %s\n会话: #%s\n", + i+1, from, subj, mid, alias) + + // 抄送 + if ccList, ok := mail["cc_list"].([]interface{}); ok && len(ccList) > 0 { + names := make([]string, 0, len(ccList)) + for _, c := range ccList { + if cc, ok := c.(map[string]interface{}); ok { + if n, ok := cc["name"].(string); ok { + names = append(names, n) + } + } + } + if len(names) > 0 { + fmt.Fprintf(&sb, "抄送: %s\n", strings.Join(names, "、")) + } + } + + // 附件 + if atts, ok := mail["attachments"].([]interface{}); ok && len(atts) > 0 { + fmt.Fprintf(&sb, "附件:\n") + for _, a := range atts { + if att, ok := a.(map[string]interface{}); ok { + fn, _ := att["filename"].(string) + sz, _ := att["size_bytes"].(float64) + aid, _ := att["attachment_id"].(string) + fmt.Fprintf(&sb, " - %s (%.1fKB, id=%s)\n", fn, sz/1024, aid) + } + } + } + + if body != "" { + if len(body) > 1000 { + body = body[:1000] + "..." + } + fmt.Fprintf(&sb, "内容: %s\n", body) + } + sb.WriteString("\n") + } + + // 标记已读 + ids := make([]string, 0, len(mails)) + for _, m := range mails { + if mail, ok := m.(map[string]interface{}); ok { + if id, ok := mail["mail_id"].(string); ok { + ids = append(ids, id) + } + } + } + if len(ids) > 0 { + go p.markRead(ids) + } + + return map[string]interface{}{ + "content": []map[string]interface{}{{"type": "text", "text": sb.String()}}, + }, nil +} + +func (p *Plugin) handleSendMail(args map[string]interface{}) (interface{}, error) { + to, _ := args["to"].(string) + subj, _ := args["subject"].(string) + body, _ := args["body"].(string) + cc, _ := args["cc"].(string) + replyTo, _ := args["reply_to"].(string) + + if to == "" || subj == "" || body == "" { + return nil, fmt.Errorf("缺少必填字段：to, subject, body") + } + + payload := map[string]interface{}{ + "to": to, + "subject": subj, + "body": body, + } + if cc != "" { + payload["cc"] = cc + } + if replyTo != "" { + payload["reply_to"] = replyTo + } + + var result map[string]interface{} + if err := p.post("/mail/send", payload, &result); err != nil { + return nil, err + } + + mid, _ := result["mail_id"].(string) + sid, _ := result["session_id"].(string) + text := fmt.Sprintf("邮件已发送（ID: %s，Session: %s）", mid, sid) + if budget, ok := result["budget_remaining"].(float64); ok { + text += fmt.Sprintf("。本任务剩余 %.0f 个来回", budget) + } + + return map[string]interface{}{ + "content": []map[string]interface{}{{"type": "text", "text": text}}, + }, nil +} + +// ─── HTTP 辅助 ─── + +func (p *Plugin) get(url string, out interface{}) error { + req, err := http.NewRequest("GET", url, nil) + if err != nil { + return err + } + req.Header.Set("Authorization", "Bearer "+p.key) + + resp, err := p.client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + if resp.StatusCode >= 400 { + body, _ := io.ReadAll(resp.Body) + return fmt.Errorf("HTTP %d: %s", resp.StatusCode, string(body)) + } + return json.NewDecoder(resp.Body).Decode(out) +} + +func (p *Plugin) post(path string, payload interface{}, out interface{}) error { + data, err := json.Marshal(payload) + if err != nil { + return err + } + + req, err := http.NewRequest("POST", p.gwURL+"/api/v1"+path, bytes.NewReader(data)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer "+p.key) + + resp, err := p.client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + if resp.StatusCode >= 400 { + body, _ := io.ReadAll(resp.Body) + return fmt.Errorf("POST %s HTTP %d: %s", path, resp.StatusCode, string(body)) + } + if out != nil { + return json.NewDecoder(resp.Body).Decode(out) + } + return nil +} + +func (p *Plugin) markRead(ids []string) { + data, _ := json.Marshal(map[string]interface{}{"mail_ids": ids}) + req, err := http.NewRequest("POST", p.gwURL+"/api/v1/mail/read", bytes.NewReader(data)) + if err != nil { + return + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer "+p.key) + p.client.Do(req) +} diff --git a/plugins/homeagent-mail-bridge/plugin.json b/plugins/homeagent-mail-bridge/plugin.json new file mode 100644 index 0000000..844f01d --- /dev/null +++ b/plugins/homeagent-mail-bridge/plugin.json @@ -0,0 +1,13 @@ +{ + "author": "AgentMail", + "description": "HomeAgent 接入 AgentMail：邮件驱动的多智能体协作", + "entry": "plugin.bin", + "name": "homeagent-mail-bridge", + "name_en": "homeagent", + "name_zh": "AgentMail 桥接", + "tags": [ + "mail", + "agentmail" + ], + "version": "0.1.0" +} \ No newline at end of file