From 6184736fd4934ba21114315c5403c4765bbe70f5 Mon Sep 17 00:00:00 2001 From: JianFeeeee Date: Wed, 26 Aug 2026 19:30:34 +0800 Subject: [PATCH] =?UTF-8?q?feat(examples):=20a2a/acp=20=E4=BC=9A=E8=AF=9D?= =?UTF-8?q?=E5=8E=86=E5=8F=B2=E6=9F=A5=E8=AF=A2=20+=20=E5=87=BA=E7=AB=99?= =?UTF-8?q?=20session=5Fid=20=E9=80=8F=E4=BC=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit a2a tasks.get/session.get 返回会话近N条消息；acp 新增 session/get； a2a_query/acp_query 接受 session_id 延续对方会话。详见 TrueAgent 仓库。 --- example/a2a/plg.json | 2 +- example/a2a/plugin.go | 93 +++++++++++++++++++++++++++++++++++++++---- example/acp/plg.json | 2 +- example/acp/plugin.go | 67 +++++++++++++++++++++++++++++-- 4 files changed, 151 insertions(+), 13 deletions(-) diff --git a/example/a2a/plg.json b/example/a2a/plg.json index f1bf996..b089e19 100644 --- a/example/a2a/plg.json +++ b/example/a2a/plg.json @@ -2,7 +2,7 @@ "name": "a2a", "name_zh": "A2A 代理通信", "name_en": "A2A Agent Communication", - "version": "1.2.0", + "version": "1.3.0", "description": "Agent-to-Agent 协议通信插件，支持双向 A2A 通信：可查询其他 Agent 并回复其请求。提供 HTTP 服务端暴露本 Agent 能力。", "author": "HomeAgent", "entry": "plugin.so", diff --git a/example/a2a/plugin.go b/example/a2a/plugin.go index f6ca462..d48214b 100644 --- a/example/a2a/plugin.go +++ b/example/a2a/plugin.go @@ -77,6 +77,7 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error { "properties": map[string]interface{}{ "agent_url": map[string]interface{}{"type": "string", "description": "目标 Agent 的 A2A 端点 URL"}, "query": map[string]interface{}{"type": "string", "description": "发送给目标 Agent 的文本查询"}, + "session_id": map[string]interface{}{"type": "string", "description": "可选。上次调用返回的 session_id，传入可延续与该 agent 的多轮对话上下文"}, "timeout": map[string]interface{}{"type": "integer", "description": "超时时间（秒），默认 60"}, }, "required": []string{"agent_url", "query"}, @@ -169,6 +170,43 @@ func truncateRunes(s string, n int) string { return string(r[:n]) + "..." } +// sessionMessages 返回指定会话的近 limit 条消息（时间正序）， +// 会话不存在返回 nil。消息格式 [{role, text, ts}]。 +func (p *Plugin) sessionMessages(sessionID string, limit int) []map[string]interface{} { + p.sessMu.Lock() + sess := p.sessions[sessionID] + var hist []string + var lastUsed time.Time + if sess != nil { + hist = append([]string{}, sess.History...) + lastUsed = sess.LastUsed + } + p.sessMu.Unlock() + if sess == nil { + return nil + } + _ = lastUsed + // History 交替 [user, agent, user, agent...]，取末尾 limit 条，保持时间正序 + start := 0 + if len(hist) > limit { + start = len(hist) - limit + } + msgs := make([]map[string]interface{}, 0, len(hist)-start) + for i := start; i < len(hist); i++ { + role, text := "user", hist[i] + if after, ok := strings.CutPrefix(text, "用户: "); ok { + role, text = "user", after + } else if after, ok := strings.CutPrefix(text, "助手: "); ok { + role, text = "agent", after + } + msgs = append(msgs, map[string]interface{}{ + "role": role, + "text": text, + }) + } + return msgs +} + func (p *Plugin) stopServer() { p.srvMu.Lock() defer p.srvMu.Unlock() @@ -248,6 +286,7 @@ func (p *Plugin) handleIncomingA2A(w http.ResponseWriter, r *http.Request) { Params struct { Query string `json:"query,omitempty"` SessionID string `json:"session_id,omitempty"` + Limit int `json:"limit,omitempty"` Message *struct { Role string `json:"role"` Parts []struct { @@ -330,11 +369,37 @@ func (p *Plugin) handleIncomingA2A(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(resp) - case "tasks.get": + case "tasks.get", "session.get": + // 按 session_id 返回会话内近 N 条消息（默认 10 条）。 + sessionID := strings.TrimSpace(req.Params.SessionID) + if sessionID == "" { + sessionID = strings.TrimSpace(req.Params.Query) + } + limit := 10 + if req.Params.Limit > 0 && req.Params.Limit <= 100 { + limit = req.Params.Limit + } + msgs := p.sessionMessages(sessionID, limit) + if msgs == nil { + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]interface{}{ + "jsonrpc": "2.0", "id": req.ID, + "result": map[string]interface{}{ + "session_id": sessionID, + "status": "not_found", + "messages": []interface{}{}, + }, + }) + return + } w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(map[string]interface{}{ "jsonrpc": "2.0", "id": req.ID, - "result": map[string]interface{}{"id": req.Params.Query, "status": "unknown"}, + "result": map[string]interface{}{ + "session_id": sessionID, + "status": "completed", + "messages": msgs, + }, }) default: @@ -378,9 +443,10 @@ type A2ARequest struct { } type A2AParams struct { - Query string `json:"query,omitempty"` - Message *A2AMessage `json:"message,omitempty"` - TaskID string `json:"id,omitempty"` + Query string `json:"query,omitempty"` + SessionID string `json:"session_id,omitempty"` + Message *A2AMessage `json:"message,omitempty"` + TaskID string `json:"id,omitempty"` } type A2AResponse struct { @@ -393,6 +459,7 @@ type A2AResponse struct { type A2AResult struct { TaskID string `json:"id,omitempty"` Status string `json:"status,omitempty"` + SessionID string `json:"session_id,omitempty"` Message *A2AMessage `json:"message,omitempty"` AgentCard *A2AAgentCard `json:"agent_card,omitempty"` } @@ -458,6 +525,7 @@ func (p *Plugin) handleA2ADiscover(args map[string]interface{}) (interface{}, er func (p *Plugin) handleA2AQuery(args map[string]interface{}) (interface{}, error) { agentURL, _ := args["agent_url"].(string) query, _ := args["query"].(string) + sessionID, _ := args["session_id"].(string) // 可选：延续对方会话 timeoutSec := 60 if v, ok := args["timeout"].(float64); ok && v > 0 { timeoutSec = int(v) @@ -479,7 +547,8 @@ func (p *Plugin) handleA2AQuery(args map[string]interface{}) (interface{}, error ID: fmt.Sprintf("a2a_%d", time.Now().UnixNano()), Method: "tasks.send", Params: A2AParams{ - Message: &A2AMessage{Role: "user", Parts: []A2APart{{Text: query, Type: "text"}}}, + SessionID: sessionID, + Message: &A2AMessage{Role: "user", Parts: []A2APart{{Text: query, Type: "text"}}}, }, } @@ -518,10 +587,18 @@ func (p *Plugin) handleA2AQuery(args map[string]interface{}) (interface{}, error replyText = strings.TrimSpace(replyText) } - return map[string]interface{}{ + result := map[string]interface{}{ "task_id": a2aResp.Result.TaskID, "status": a2aResp.Result.Status, "response": replyText, - }, nil + } + if a2aResp.Result.SessionID != "" || sessionID != "" { + result["session_id"] = a2aResp.Result.SessionID + if result["session_id"] == "" { + result["session_id"] = sessionID + } + result["note"] = "延续会话：下次调用传此 session_id 可保持上下文" + } + return result, nil } // ---- Management Handlers ---- diff --git a/example/acp/plg.json b/example/acp/plg.json index f67c8b8..e9fe3e1 100644 --- a/example/acp/plg.json +++ b/example/acp/plg.json @@ -2,7 +2,7 @@ "name": "acp", "name_zh": "ACP 代理通信", "name_en": "ACP Agent Client Protocol", - "version": "1.1.0", + "version": "1.2.0", "description": "Agent Client Protocol 通信插件：充当 ACP 服务端接受其他 Agent 的任务请求，同时提供客户端工具向远程 ACP Agent（如 opencode）发起会话并读取回复", "author": "HomeAgent", "entry": "plugin.so", diff --git a/example/acp/plugin.go b/example/acp/plugin.go index 7c91bc7..72dc373 100644 --- a/example/acp/plugin.go +++ b/example/acp/plugin.go @@ -71,6 +71,7 @@ func (p *Plugin) Start(s *sdk.PluginSDK) error { "properties": map[string]interface{}{ "server_url": map[string]interface{}{"type": "string", "description": "目标 ACP 服务端地址（如 http://127.0.0.1:13000）"}, "prompt": map[string]interface{}{"type": "string", "description": "发送给目标 Agent 的任务描述"}, + "session_id": map[string]interface{}{"type": "string", "description": "可选。上次调用返回的 session_id，传入可延续与该 agent 的多轮对话上下文"}, "timeout": map[string]interface{}{"type": "integer", "description": "等待回复超时（秒），默认 120"}, }, "required": []string{"server_url", "prompt"}, @@ -184,6 +185,7 @@ func (p *Plugin) handleSessionPost(w http.ResponseWriter, r *http.Request) { Text string `json:"text"` } `json:"request,omitempty"` SessionID string `json:"session_id,omitempty"` + Limit int `json:"limit,omitempty"` Final bool `json:"final,omitempty"` } `json:"params,omitempty"` } @@ -256,6 +258,59 @@ func (p *Plugin) handleSessionPost(w http.ResponseWriter, r *http.Request) { }, }) + case "session/get": + // 按 session_id 返回会话内近 N 条消息（默认 10 条，时间正序） + sid := req.Params.SessionID + p.mu.RLock() + st := p.sessions[sid] + var hist []string + if st != nil { + hist = append([]string{}, st.History...) + } + p.mu.RUnlock() + if st == nil { + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]interface{}{ + "jsonrpc": "2.0", "id": req.ID, + "result": map[string]interface{}{ + "session_id": sid, + "status": "not_found", + "messages": []interface{}{}, + }, + }) + return + } + limit := 10 + if req.Params.Limit > 0 && req.Params.Limit <= 100 { + limit = req.Params.Limit + } + start := 0 + if len(hist) > limit { + start = len(hist) - limit + } + msgs := make([]map[string]interface{}, 0, len(hist)-start) + for i := start; i < len(hist); i++ { + role, text := "user", hist[i] + if after, ok := strings.CutPrefix(text, "用户: "); ok { + role, text = "user", after + } else if after, ok := strings.CutPrefix(text, "助手: "); ok { + role, text = "agent", after + } + msgs = append(msgs, map[string]interface{}{ + "role": role, + "text": text, + }) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]interface{}{ + "jsonrpc": "2.0", "id": req.ID, + "result": map[string]interface{}{ + "session_id": sid, + "status": "completed", + "messages": msgs, + }, + }) + case "session/update": sid := req.Params.SessionID p.mu.Lock() @@ -391,6 +446,7 @@ func (p *Plugin) handleAcpQuery(args map[string]interface{}) (interface{}, error if prompt == "" { return map[string]interface{}{"error": "prompt 不能为空"}, nil } + sessionID, _ := args["session_id"].(string) // 可选：延续对方会话 timeoutSec := 120 if v, ok := args["timeout"].(float64); ok && v > 0 { timeoutSec = int(v) @@ -399,12 +455,16 @@ func (p *Plugin) handleAcpQuery(args map[string]interface{}) (interface{}, error endpoint := serverURL + "/api/session" client := &http.Client{Timeout: time.Duration(timeoutSec) * time.Second} + params := map[string]interface{}{ + "request": map[string]interface{}{"text": prompt}, + } + if sessionID != "" { + params["session_id"] = sessionID + } newBody, _ := json.Marshal(map[string]interface{}{ "jsonrpc": "2.0", "id": "acp-" + fmt.Sprintf("%d", time.Now().UnixNano()), "method": "session/new", - "params": map[string]interface{}{ - "request": map[string]interface{}{"text": prompt}, - }, + "params": params, }) req, _ := http.NewRequest("POST", endpoint, bytes.NewReader(newBody)) @@ -469,6 +529,7 @@ func (p *Plugin) handleAcpQuery(args map[string]interface{}) (interface{}, error "session_id": sid, "status": "completed", "reply": replyText, + "note": "延续会话：下次调用传此 session_id 可保持上下文", }, nil }