From 3b34ba78b8498669df117dc38a57fd212a6557b3 Mon Sep 17 00:00:00 2001 From: QihuAI Design Date: Sun, 19 Jul 2026 15:27:15 +0800 Subject: [PATCH] =?UTF-8?q?feat(agents):=20Phase3.2=20Agent=E5=87=AD?= =?UTF-8?q?=E6=8D=AEFernet=E5=8A=A0=E5=AF=86=E5=AD=98=E5=82=A8=20=E5=86=99?= =?UTF-8?q?=E5=85=A5=E5=8A=A0=E5=AF=86+=E8=B0=83=E7=94=A8=E7=82=B9?= =?UTF-8?q?=E8=A7=A3=E5=AF=86+=E8=BF=94=E5=9B=9E=E8=84=B1=E6=95=8F+?= =?UTF-8?q?=E4=B8=BB=E5=AF=86=E9=92=A5=E6=8C=81=E4=B9=85=E5=8C=96=E9=98=B2?= =?UTF-8?q?=E7=AB=9E=E6=80=81=20[v2026.07.19.7]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/python/app/api/chat_agents.py | 1894 ++++++++++++------------ backend/python/app/api/chat_skill.py | 3 +- backend/python/app/api/market.py | 3 +- backend/python/app/core/credential.py | 105 ++ backend/python/app/models.py | 3 +- migrations/023_widen_agent_api_key.sql | 11 + 6 files changed, 1073 insertions(+), 946 deletions(-) create mode 100644 backend/python/app/core/credential.py create mode 100644 migrations/023_widen_agent_api_key.sql diff --git a/backend/python/app/api/chat_agents.py b/backend/python/app/api/chat_agents.py index b843bda..6ea17e3 100644 --- a/backend/python/app/api/chat_agents.py +++ b/backend/python/app/api/chat_agents.py @@ -1,943 +1,951 @@ -# Auto-extracted from chat.py during Phase 1 refactor (Strangler Fig). -# Source of truth: scripts/phase1_extract.py. Do not hand-merge back. - -from fastapi import APIRouter, Depends, HTTPException, Query, Response -from sqlalchemy.orm import Session -from sqlalchemy import and_, desc, asc, func -from pydantic import BaseModel, Field -from typing import Optional -import uuid -from datetime import datetime, timezone, timedelta -import re -import json -import asyncio -import math -import httpx -import os -import logging -from ..database import get_db, SessionLocal -from ..models import ( - - - - User, Conversation, Message, AIModel, ModelProvider, TokenUsageLog, - - - - KnowledgeBase, KnowledgeDocument, KnowledgeDocumentChunk, EvolutionSkill, - - - - EvolutionTrajectory, MarketApp, AppPurchase, AppInstallation, - - - - MCPServer, - - - - gen_uuid - - - -) -from ..auth import get_current_user -from ..platform_knowledge import get_platform_knowledge -from ..mcp_client import list_mcp_tools, call_mcp_tool -from ..middleware.log_service import LogService -from ..capability_registry import resolve_auto_model -from app.api.chat_utils import (_extract_text_from_json, _proxy_for_url) - -def _normalize_dify_chat_url(endpoint: str) -> str: - - - - """把用户填写的 Dify 地址归一化为 /v1/chat-messages 端点。与 market.py 保持一致。""" - - - - ep = (endpoint or "").rstrip("/") - - - - if ep.endswith("/chat-messages"): - - - - return ep - - - - if ep.endswith("/v1"): - - - - return ep + "/chat-messages" - - - - return ep + "/v1/chat-messages" - -async def _call_dify_agent(endpoint: str, api_key: str, query: str, - - - - user_id: str, ext_conv_id: str | None) -> tuple: - - - - """调用 Dify(新版 Agent) /v1/chat-messages，流式(SSE)解析最终回答。 - - - - - - - - 新版 Agent API 仅支持 response_mode=streaming(blocking 会 400)。事件流中 - - - - message / agent_message 事件携带增量 answer，累加即为最终回答；每个事件都带 - - - - conversation_id，用于维持多轮上下文。返回 (answer, new_conversation_id)。 - - - - """ - - - - url = _normalize_dify_chat_url(endpoint) - - - - headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} - - - - payload = { - - - - "inputs": {}, - - - - "query": query, - - - - "user": (user_id or "web-user")[:64], - - - - "response_mode": "streaming", - - - - "conversation_id": ext_conv_id or "", - - - - } - - - - answer_parts = [] - - - - new_conv_id = ext_conv_id - - - - err_msg = None - - - - wf_outputs = None - - - - # httpx 默认 follow_redirects=False；stream POST 若遇 301/302 会直接返回该状态码。 - - - - # 且 httpx 自动跟随会把 POST 降级为 GET(对 /v1/chat-messages 变 405)。 - - - - # 因此手动跟随 301/302/307/308，保持 POST 与请求体不变(最多 4 跳)。 - - - - async with httpx.AsyncClient(timeout=180, proxy=_proxy_for_url(url), trust_env=False, - - - - follow_redirects=False) as client: - - - - cur_url = url - - - - for _redir in range(4): - - - - async with client.stream("POST", cur_url, headers=headers, json=payload) as resp: - - - - if resp.status_code in (301, 302, 307, 308): - - - - loc = resp.headers.get("location") or resp.headers.get("Location") - - - - await resp.aclose() - - - - if not loc: - - - - raise RuntimeError(f"Dify HTTP {resp.status_code}: 缺少重定向地址") - - - - # 处理相对地址 - - - - if loc.startswith("/"): - - - - from urllib.parse import urlsplit, urlunsplit - - - - p = urlsplit(cur_url) - - - - loc = urlunsplit((p.scheme, p.netloc, loc, "", "")) - - - - cur_url = loc - - - - continue - - - - if resp.status_code != 200: - - - - body = (await resp.aread()).decode("utf-8", "replace") - - - - raise RuntimeError(f"Dify HTTP {resp.status_code}: {body[:300]}") - - - - async for line in resp.aiter_lines(): - - - - if not line: - - - - continue - - - - line = line.strip() - - - - if not line.startswith("data:"): - - - - continue - - - - data_str = line[5:].strip() - - - - if not data_str or data_str == "[DONE]": - - - - continue - - - - try: - - - - evt = json.loads(data_str) - - - - except Exception: - - - - continue - - - - ev = evt.get("event") - - - - if evt.get("conversation_id"): - - - - new_conv_id = evt["conversation_id"] - - - - data = evt.get("data") or {} - - - - # 兼容三类 Dify 应用： - - - - # 1) Chat/Agent 应用: message / agent_message 事件带增量 answer - - - - # 2) Chatflow/Workflow 应用: text_chunk 事件带 data.text； - - - - # 结束时 workflow_finished 带 status / outputs / error - - - - if ev in ("message", "agent_message"): - - - - answer_parts.append(evt.get("answer", "") or "") - - - - elif ev == "message_replace": - - - - # 内容审查替换：以替换后的完整 answer 为准 - - - - answer_parts = [evt.get("answer", "") or ""] - - - - elif ev == "text_chunk": - - - - answer_parts.append((data.get("text") or "")) - - - - elif ev == "node_finished": - - - - # answer / end 节点的 outputs 可能含最终文本 - - - - if data.get("node_type") in ("answer", "end") and data.get("outputs"): - - - - wf_outputs = data.get("outputs") - - - - if data.get("status") == "failed" and data.get("error"): - - - - err_msg = str(data.get("error")) - - - - elif ev == "workflow_finished": - - - - if data.get("status") == "failed": - - - - err_msg = str(data.get("error") or "Dify 工作流执行失败") - - - - elif data.get("outputs"): - - - - wf_outputs = data.get("outputs") - - - - elif ev == "error": - - - - err_msg = evt.get("message") or evt.get("code") or "Dify 流式返回错误" - - - - break - - - - else: - - - - raise RuntimeError("Dify 重定向次数过多(>4)，请直接填写重定向后的地址") - - - - answer = "".join(answer_parts).strip() - - - - # 流式无增量文本时，尝试从 workflow 结束 outputs 兜底提取 - - - - if not answer and wf_outputs: - - - - answer = _extract_text_from_json(wf_outputs).strip() - - - - if not answer and err_msg: - - - - raise RuntimeError(err_msg) - - - - return answer, new_conv_id - -async def _call_n8n_agent(endpoint: str, api_key: str, query: str, - - - - user_id: str, history: list) -> tuple: - - - - """调用 n8n 生产 Webhook (POST)。webhook 用 Respond to Webhook 节点返回 JSON， - - - - 尽力从中抽取文本回答。n8n webhook 无平台级会话态，多轮靠我们下发 history。 - - - - 返回 (answer, None)。""" - - - - url = (endpoint or "").strip() - - - - headers = {"Content-Type": "application/json"} - - - - if api_key: - - - - # n8n Header Auth 自定义头名不固定，同时下发常见几种以提高兼容性 - - - - headers["Authorization"] = f"Bearer {api_key}" - - - - headers["X-N8N-API-KEY"] = api_key - - - - # 携带多种字段别名，兼容 n8n AI Agent(chatInput) 与自定义工作流 - - - - payload = { - - - - "query": query, - - - - "chatInput": query, - - - - "message": query, - - - - "input": query, - - - - "user": (user_id or "web-user")[:64], - - - - "history": history or [], - - - - } - - - - async with httpx.AsyncClient(timeout=180, proxy=_proxy_for_url(url), trust_env=False) as client: - - - - resp = await client.post(url, headers=headers, json=payload) - - - - if resp.status_code not in (200, 201): - - - - raise RuntimeError(f"n8n Webhook HTTP {resp.status_code}: {resp.text[:300]}") - - - - ctype = resp.headers.get("content-type", "") - - - - if "application/json" in ctype: - - - - try: - - - - answer = _extract_text_from_json(resp.json()) - - - - except Exception: - - - - answer = resp.text - - - - else: - - - - answer = resp.text - - - - return (answer or "").strip(), None - -async def _call_coze_agent(endpoint: str, api_key: str, query: str, - - - - user_id: str, ext_conv_id: str | None) -> tuple: - - - - """调用扣子(Coze) v3 对话 API：create → 轮询 retrieve → 拉取 message list。 - - - - - - - - 约定 agent_endpoint 存放 bot_id(纯 id)，或形如 - - - - https://api.coze.cn/v3/chat?bot_id=xxx 的完整地址。默认基址 api.coze.cn。 - - - - 返回 (answer, conversation_id)。 - - - - """ - - - - ep = (endpoint or "").strip() - - - - from urllib.parse import urlparse, parse_qs - - - - bot_id = None - - - - base = "https://api.coze.cn" - - - - if ep.startswith("http"): - - - - pr = urlparse(ep) - - - - base = f"{pr.scheme}://{pr.netloc}" - - - - qs = parse_qs(pr.query) - - - - bot_id = (qs.get("bot_id") or [None])[0] - - - - if not bot_id and "/bot/" in pr.path: - - - - bot_id = pr.path.rstrip("/").split("/bot/")[-1] - - - - else: - - - - bot_id = ep # 纯 bot_id - - - - if not bot_id: - - - - raise RuntimeError("Coze 接入缺少 bot_id(请在 Endpoint 填入 bot_id 或带 ?bot_id= 的地址)") - - - - headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} - - - - create_url = f"{base}/v3/chat" - - - - if ext_conv_id: - - - - create_url += f"?conversation_id={ext_conv_id}" - - - - payload = { - - - - "bot_id": bot_id, - - - - "user_id": (user_id or "web-user")[:64], - - - - "stream": False, - - - - "auto_save_history": True, - - - - "additional_messages": [ - - - - {"role": "user", "content": query, "content_type": "text"} - - - - ], - - - - } - - - - async with httpx.AsyncClient(timeout=180, proxy=_proxy_for_url(create_url), trust_env=False) as client: - - - - r = await client.post(create_url, headers=headers, json=payload) - - - - if r.status_code != 200: - - - - raise RuntimeError(f"Coze create HTTP {r.status_code}: {r.text[:300]}") - - - - cd = r.json() - - - - if cd.get("code") not in (0, None): - - - - raise RuntimeError(f"Coze 错误: {cd.get('msg') or cd}") - - - - chat = cd.get("data") or {} - - - - chat_id = chat.get("id") - - - - conv_id2 = chat.get("conversation_id") or ext_conv_id - - - - if not chat_id or not conv_id2: - - - - raise RuntimeError(f"Coze 未返回 chat_id/conversation_id: {str(cd)[:200]}") - - - - # 轮询直到完成(最多约 3 分钟) - - - - status = chat.get("status") - - - - for _ in range(60): - - - - if status == "completed": - - - - break - - - - if status in ("failed", "requires_action", "canceled"): - - - - raise RuntimeError(f"Coze 对话状态 {status}") - - - - await asyncio.sleep(3) - - - - rr = await client.get( - - - - f"{base}/v3/chat/retrieve?chat_id={chat_id}&conversation_id={conv_id2}", - - - - headers=headers, - - - - ) - - - - if rr.status_code == 200: - - - - status = ((rr.json() or {}).get("data") or {}).get("status") - - - - # 拉取消息列表，取 type=answer 的内容 - - - - rm = await client.get( - - - - f"{base}/v3/chat/message/list?chat_id={chat_id}&conversation_id={conv_id2}", - - - - headers=headers, - - - - ) - - - - answer = "" - - - - if rm.status_code == 200: - - - - for m in ((rm.json() or {}).get("data") or []): - - - - if m.get("type") == "answer" and m.get("role") == "assistant": - - - - answer = m.get("content", "") or answer - - - - return (answer or "").strip(), conv_id2 - -async def _call_external_agent(db, agent: dict, query: str, user_id: str, - - - - ext_conversation_id: str | None, - - - - history: list) -> tuple: - - - - """按平台分发到对应外部 Agent 远程调用。返回 (answer, new_ext_conversation_id)。""" - - - - platform = (agent.get("platform") or "dify").lower() - - - - endpoint = agent.get("agent_endpoint") - - - - api_key = agent.get("agent_api_key") or "" - - - - if not endpoint: - - - - raise RuntimeError("该 Agent 未配置远程调用地址(agent_endpoint)") - - - - if platform == "dify": - - - - return await _call_dify_agent(endpoint, api_key, query, user_id, ext_conversation_id) - - - - if platform == "n8n": - - - - return await _call_n8n_agent(endpoint, api_key, query, user_id, history) - - - - if platform == "coze": - - - - return await _call_coze_agent(endpoint, api_key, query, user_id, ext_conversation_id) - - - - # 未知平台：默认按 Dify 协议尝试 - - - - return await _call_dify_agent(endpoint, api_key, query, user_id, ext_conversation_id) +# Auto-extracted from chat.py during Phase 1 refactor (Strangler Fig). +# Source of truth: scripts/phase1_extract.py. Do not hand-merge back. + +from app.core.credential import decrypt_secret +from app.core.observability import get_trace_id +from fastapi import APIRouter, Depends, HTTPException, Query, Response +from sqlalchemy.orm import Session +from sqlalchemy import and_, desc, asc, func +from pydantic import BaseModel, Field +from typing import Optional +import uuid +from datetime import datetime, timezone, timedelta +import re +import json +import asyncio +import math +import httpx +import os +import logging +from ..database import get_db, SessionLocal +from ..models import ( + + + + User, Conversation, Message, AIModel, ModelProvider, TokenUsageLog, + + + + KnowledgeBase, KnowledgeDocument, KnowledgeDocumentChunk, EvolutionSkill, + + + + EvolutionTrajectory, MarketApp, AppPurchase, AppInstallation, + + + + MCPServer, + + + + gen_uuid + + + +) +from ..auth import get_current_user +from ..platform_knowledge import get_platform_knowledge +from ..mcp_client import list_mcp_tools, call_mcp_tool +from ..middleware.log_service import LogService +from ..capability_registry import resolve_auto_model +from app.api.chat_utils import (_extract_text_from_json, _proxy_for_url) + +def _normalize_dify_chat_url(endpoint: str) -> str: + + + + """把用户填写的 Dify 地址归一化为 /v1/chat-messages 端点。与 market.py 保持一致。""" + + + + ep = (endpoint or "").rstrip("/") + + + + if ep.endswith("/chat-messages"): + + + + return ep + + + + if ep.endswith("/v1"): + + + + return ep + "/chat-messages" + + + + return ep + "/v1/chat-messages" + +async def _call_dify_agent(endpoint: str, api_key: str, query: str, + + + + user_id: str, ext_conv_id: str | None) -> tuple: + + + + """调用 Dify(新版 Agent) /v1/chat-messages，流式(SSE)解析最终回答。 + + + + + + + + 新版 Agent API 仅支持 response_mode=streaming(blocking 会 400)。事件流中 + + + + message / agent_message 事件携带增量 answer，累加即为最终回答；每个事件都带 + + + + conversation_id，用于维持多轮上下文。返回 (answer, new_conversation_id)。 + + + + """ + + + + url = _normalize_dify_chat_url(endpoint) + + + + headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} + + + + payload = { + + + + "inputs": {}, + + + + "query": query, + + + + "user": (user_id or "web-user")[:64], + + + + "response_mode": "streaming", + + + + "conversation_id": ext_conv_id or "", + + + + } + + + + answer_parts = [] + + + + new_conv_id = ext_conv_id + + + + err_msg = None + + + + wf_outputs = None + + + + # httpx 默认 follow_redirects=False；stream POST 若遇 301/302 会直接返回该状态码。 + + + + # 且 httpx 自动跟随会把 POST 降级为 GET(对 /v1/chat-messages 变 405)。 + + + + # 因此手动跟随 301/302/307/308，保持 POST 与请求体不变(最多 4 跳)。 + + + + async with httpx.AsyncClient(timeout=180, proxy=_proxy_for_url(url), trust_env=False, + + + + follow_redirects=False) as client: + + + + cur_url = url + + + + for _redir in range(4): + + + + async with client.stream("POST", cur_url, headers=headers, json=payload) as resp: + + + + if resp.status_code in (301, 302, 307, 308): + + + + loc = resp.headers.get("location") or resp.headers.get("Location") + + + + await resp.aclose() + + + + if not loc: + + + + raise RuntimeError(f"Dify HTTP {resp.status_code}: 缺少重定向地址") + + + + # 处理相对地址 + + + + if loc.startswith("/"): + + + + from urllib.parse import urlsplit, urlunsplit + + + + p = urlsplit(cur_url) + + + + loc = urlunsplit((p.scheme, p.netloc, loc, "", "")) + + + + cur_url = loc + + + + continue + + + + if resp.status_code != 200: + + + + body = (await resp.aread()).decode("utf-8", "replace") + + + + raise RuntimeError(f"Dify HTTP {resp.status_code}: {body[:300]}") + + + + async for line in resp.aiter_lines(): + + + + if not line: + + + + continue + + + + line = line.strip() + + + + if not line.startswith("data:"): + + + + continue + + + + data_str = line[5:].strip() + + + + if not data_str or data_str == "[DONE]": + + + + continue + + + + try: + + + + evt = json.loads(data_str) + + + + except Exception: + + + + continue + + + + ev = evt.get("event") + + + + if evt.get("conversation_id"): + + + + new_conv_id = evt["conversation_id"] + + + + data = evt.get("data") or {} + + + + # 兼容三类 Dify 应用： + + + + # 1) Chat/Agent 应用: message / agent_message 事件带增量 answer + + + + # 2) Chatflow/Workflow 应用: text_chunk 事件带 data.text； + + + + # 结束时 workflow_finished 带 status / outputs / error + + + + if ev in ("message", "agent_message"): + + + + answer_parts.append(evt.get("answer", "") or "") + + + + elif ev == "message_replace": + + + + # 内容审查替换：以替换后的完整 answer 为准 + + + + answer_parts = [evt.get("answer", "") or ""] + + + + elif ev == "text_chunk": + + + + answer_parts.append((data.get("text") or "")) + + + + elif ev == "node_finished": + + + + # answer / end 节点的 outputs 可能含最终文本 + + + + if data.get("node_type") in ("answer", "end") and data.get("outputs"): + + + + wf_outputs = data.get("outputs") + + + + if data.get("status") == "failed" and data.get("error"): + + + + err_msg = str(data.get("error")) + + + + elif ev == "workflow_finished": + + + + if data.get("status") == "failed": + + + + err_msg = str(data.get("error") or "Dify 工作流执行失败") + + + + elif data.get("outputs"): + + + + wf_outputs = data.get("outputs") + + + + elif ev == "error": + + + + err_msg = evt.get("message") or evt.get("code") or "Dify 流式返回错误" + + + + break + + + + else: + + + + raise RuntimeError("Dify 重定向次数过多(>4)，请直接填写重定向后的地址") + + + + answer = "".join(answer_parts).strip() + + + + # 流式无增量文本时，尝试从 workflow 结束 outputs 兜底提取 + + + + if not answer and wf_outputs: + + + + answer = _extract_text_from_json(wf_outputs).strip() + + + + if not answer and err_msg: + + + + raise RuntimeError(err_msg) + + + + return answer, new_conv_id + +async def _call_n8n_agent(endpoint: str, api_key: str, query: str, + + + + user_id: str, history: list) -> tuple: + + + + """调用 n8n 生产 Webhook (POST)。webhook 用 Respond to Webhook 节点返回 JSON， + + + + 尽力从中抽取文本回答。n8n webhook 无平台级会话态，多轮靠我们下发 history。 + + + + 返回 (answer, None)。""" + + + + url = (endpoint or "").strip() + + + + headers = {"Content-Type": "application/json"} + + + + if api_key: + + + + # n8n Header Auth 自定义头名不固定，同时下发常见几种以提高兼容性 + + + + headers["Authorization"] = f"Bearer {api_key}" + + + + headers["X-N8N-API-KEY"] = api_key + + + + # 携带多种字段别名，兼容 n8n AI Agent(chatInput) 与自定义工作流 + + + + payload = { + + + + "query": query, + + + + "chatInput": query, + + + + "message": query, + + + + "input": query, + + + + "user": (user_id or "web-user")[:64], + + + + "history": history or [], + + + + } + + + + async with httpx.AsyncClient(timeout=180, proxy=_proxy_for_url(url), trust_env=False) as client: + + + + resp = await client.post(url, headers=headers, json=payload) + + + + if resp.status_code not in (200, 201): + + + + raise RuntimeError(f"n8n Webhook HTTP {resp.status_code}: {resp.text[:300]}") + + + + ctype = resp.headers.get("content-type", "") + + + + if "application/json" in ctype: + + + + try: + + + + answer = _extract_text_from_json(resp.json()) + + + + except Exception: + + + + answer = resp.text + + + + else: + + + + answer = resp.text + + + + return (answer or "").strip(), None + +async def _call_coze_agent(endpoint: str, api_key: str, query: str, + + + + user_id: str, ext_conv_id: str | None) -> tuple: + + + + """调用扣子(Coze) v3 对话 API：create → 轮询 retrieve → 拉取 message list。 + + + + + + + + 约定 agent_endpoint 存放 bot_id(纯 id)，或形如 + + + + https://api.coze.cn/v3/chat?bot_id=xxx 的完整地址。默认基址 api.coze.cn。 + + + + 返回 (answer, conversation_id)。 + + + + """ + + + + ep = (endpoint or "").strip() + + + + from urllib.parse import urlparse, parse_qs + + + + bot_id = None + + + + base = "https://api.coze.cn" + + + + if ep.startswith("http"): + + + + pr = urlparse(ep) + + + + base = f"{pr.scheme}://{pr.netloc}" + + + + qs = parse_qs(pr.query) + + + + bot_id = (qs.get("bot_id") or [None])[0] + + + + if not bot_id and "/bot/" in pr.path: + + + + bot_id = pr.path.rstrip("/").split("/bot/")[-1] + + + + else: + + + + bot_id = ep # 纯 bot_id + + + + if not bot_id: + + + + raise RuntimeError("Coze 接入缺少 bot_id(请在 Endpoint 填入 bot_id 或带 ?bot_id= 的地址)") + + + + headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} + + + + create_url = f"{base}/v3/chat" + + + + if ext_conv_id: + + + + create_url += f"?conversation_id={ext_conv_id}" + + + + payload = { + + + + "bot_id": bot_id, + + + + "user_id": (user_id or "web-user")[:64], + + + + "stream": False, + + + + "auto_save_history": True, + + + + "additional_messages": [ + + + + {"role": "user", "content": query, "content_type": "text"} + + + + ], + + + + } + + + + async with httpx.AsyncClient(timeout=180, proxy=_proxy_for_url(create_url), trust_env=False) as client: + + + + r = await client.post(create_url, headers=headers, json=payload) + + + + if r.status_code != 200: + + + + raise RuntimeError(f"Coze create HTTP {r.status_code}: {r.text[:300]}") + + + + cd = r.json() + + + + if cd.get("code") not in (0, None): + + + + raise RuntimeError(f"Coze 错误: {cd.get('msg') or cd}") + + + + chat = cd.get("data") or {} + + + + chat_id = chat.get("id") + + + + conv_id2 = chat.get("conversation_id") or ext_conv_id + + + + if not chat_id or not conv_id2: + + + + raise RuntimeError(f"Coze 未返回 chat_id/conversation_id: {str(cd)[:200]}") + + + + # 轮询直到完成(最多约 3 分钟) + + + + status = chat.get("status") + + + + for _ in range(60): + + + + if status == "completed": + + + + break + + + + if status in ("failed", "requires_action", "canceled"): + + + + raise RuntimeError(f"Coze 对话状态 {status}") + + + + await asyncio.sleep(3) + + + + rr = await client.get( + + + + f"{base}/v3/chat/retrieve?chat_id={chat_id}&conversation_id={conv_id2}", + + + + headers=headers, + + + + ) + + + + if rr.status_code == 200: + + + + status = ((rr.json() or {}).get("data") or {}).get("status") + + + + # 拉取消息列表，取 type=answer 的内容 + + + + rm = await client.get( + + + + f"{base}/v3/chat/message/list?chat_id={chat_id}&conversation_id={conv_id2}", + + + + headers=headers, + + + + ) + + + + answer = "" + + + + if rm.status_code == 200: + + + + for m in ((rm.json() or {}).get("data") or []): + + + + if m.get("type") == "answer" and m.get("role") == "assistant": + + + + answer = m.get("content", "") or answer + + + + return (answer or "").strip(), conv_id2 + +async def _call_external_agent(db, agent: dict, query: str, user_id: str, + + + + ext_conversation_id: str | None, + + + + history: list) -> tuple: + + + + """按平台分发到对应外部 Agent 远程调用。返回 (answer, new_ext_conversation_id)。""" + + + + platform = (agent.get("platform") or "dify").lower() + + + + endpoint = agent.get("agent_endpoint") + + + + _row = db.query(MarketApp).filter(MarketApp.id == agent.get("id")).first() + _raw = _row.agent_api_key if _row else None + try: + api_key = decrypt_secret(_raw) or "" + except Exception as _e: + api_key = "" + logging.error("[credential] 解密 Agent 凭据失败 id=%s trace=%s: %s" % (agent.get("id"), get_trace_id(), _e)) + + + + if not endpoint: + + + + raise RuntimeError("该 Agent 未配置远程调用地址(agent_endpoint)") + + + + if platform == "dify": + + + + return await _call_dify_agent(endpoint, api_key, query, user_id, ext_conversation_id) + + + + if platform == "n8n": + + + + return await _call_n8n_agent(endpoint, api_key, query, user_id, history) + + + + if platform == "coze": + + + + return await _call_coze_agent(endpoint, api_key, query, user_id, ext_conversation_id) + + + + # 未知平台：默认按 Dify 协议尝试 + + + + return await _call_dify_agent(endpoint, api_key, query, user_id, ext_conversation_id) diff --git a/backend/python/app/api/chat_skill.py b/backend/python/app/api/chat_skill.py index 49eed9f..d422ad0 100644 --- a/backend/python/app/api/chat_skill.py +++ b/backend/python/app/api/chat_skill.py @@ -1,6 +1,7 @@ # Auto-extracted from chat.py during Phase 1 refactor (Strangler Fig). # Source of truth: scripts/phase1_extract.py. Do not hand-merge back. +from app.core.credential import decrypt_secret, mask_secret from fastapi import APIRouter, Depends, HTTPException, Query, Response from sqlalchemy.orm import Session from sqlalchemy import and_, desc, asc, func @@ -530,7 +531,7 @@ def _resolve_market_agents(db: Session, agent_ids: list, user_id: str) -> list: - "agent_api_key": ag.agent_api_key or None, + "agent_api_key": mask_secret(decrypt_secret(ag.agent_api_key)) or None, diff --git a/backend/python/app/api/market.py b/backend/python/app/api/market.py index 723cb4b..7089a40 100644 --- a/backend/python/app/api/market.py +++ b/backend/python/app/api/market.py @@ -1,3 +1,4 @@ +from app.core.credential import encrypt_secret from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy.orm import Session from sqlalchemy import and_, desc, asc, func @@ -557,7 +558,7 @@ def upload_app( is_member_free=req.is_member_free if req.is_member_free is not None else False, join_promotion=req.join_promotion if req.join_promotion is not None else False, agent_endpoint=req.agent_endpoint, - agent_api_key=req.api_key, + agent_api_key=encrypt_secret(req.api_key), status="published", # 用户创建的应用直接发布 created_at=datetime.now(), updated_at=datetime.now(), diff --git a/backend/python/app/core/credential.py b/backend/python/app/core/credential.py new file mode 100644 index 0000000..4024296 --- /dev/null +++ b/backend/python/app/core/credential.py @@ -0,0 +1,105 @@ +# -*- coding: utf-8 -*- +"""Phase 3.2 - Agent 凭据加解密 (Fernet 对称加密)。 + +主密钥获取策略 (防竞态): +- 优先环境变量 AGENT_MASTER_KEY。 +- 否则读写 /app/secrets/agent_master_key: + * 部署阶段由 patch 显式生成并落盘 (原子写), 容器重启/提交后稳定存在。 + * 运行时只读, 文件存在即读取, 绝不重新生成 —— 避免多进程/多请求并发 + 重建密钥导致新旧密文无法互解。 + * 仅当文件确实不存在时, 才在文件锁保护下生成 (双重检查, 原子替换)。 +- 存储格式: 密文以 "enc:" 前缀标识; 非 "enc:" 视为旧明文 (向后兼容, 解密原样返回)。 +""" +import os +import logging +import fcntl +from cryptography.fernet import Fernet + +logger = logging.getLogger(__name__) + +_MASTER_KEY_FILE = "/app/secrets/agent_master_key" +_fernet_instance = None + + +def _generate_and_persist_key() -> bytes: + key = Fernet.generate_key() + os.makedirs(os.path.dirname(_MASTER_KEY_FILE), exist_ok=True) + tmp = _MASTER_KEY_FILE + ".tmp" + with open(tmp, "w") as f: + f.write(key.decode()) + f.flush() + os.fsync(f.fileno()) + try: + os.chmod(tmp, 0o600) + except Exception: + pass + os.replace(tmp, _MASTER_KEY_FILE) # 原子替换, 避免读到半截文件 + return key + + +def get_master_key() -> bytes: + env_key = os.environ.get("AGENT_MASTER_KEY") + if env_key: + return env_key.encode() if isinstance(env_key, str) else env_key + # 运行时只读: 文件已存在则直接读取 (部署阶段已生成) + if os.path.exists(_MASTER_KEY_FILE): + try: + with open(_MASTER_KEY_FILE, "r") as f: + k = f.read().strip() + if k: + return k.encode() + except Exception as e: + logger.warning("[credential] 读取主密钥失败: %s" % e) + # 文件缺失: 加锁串行生成, 防多进程并发重建 + lockf = open(_MASTER_KEY_FILE + ".lock", "w") + try: + fcntl.flock(lockf, fcntl.LOCK_EX) + if os.path.exists(_MASTER_KEY_FILE): # 双重检查 + with open(_MASTER_KEY_FILE, "r") as f: + k = f.read().strip() + if k: + return k.encode() + return _generate_and_persist_key() + finally: + fcntl.flock(lockf, fcntl.LOCK_UN) + try: + lockf.close() + except Exception: + pass + + +def _fernet() -> Fernet: + global _fernet_instance + if _fernet_instance is None: + _fernet_instance = Fernet(get_master_key()) + return _fernet_instance + + +def encrypt_secret(plain): + if plain is None or plain == "": + return None + if isinstance(plain, str) and plain.startswith("enc:"): + return plain # 已是密文, 幂等 + try: + return "enc:" + _fernet().encrypt(plain.encode()).decode() + except Exception as e: + logger.error("[credential] 加密失败: %s" % e) + return None + + +def decrypt_secret(cipher): + if cipher is None or cipher == "": + return None + if isinstance(cipher, str) and cipher.startswith("enc:"): + try: + return _fernet().decrypt(cipher[4:].encode()).decode() + except Exception as e: + logger.error("[credential] 解密失败(密钥不匹配或数据损坏): %s" % e) + return None + return cipher # 旧明文, 向后兼容 + + +def mask_secret(plain): + if not plain: + return "" + return ("*" * 8) + plain[-4:] diff --git a/backend/python/app/models.py b/backend/python/app/models.py index dab55d8..1130b45 100644 --- a/backend/python/app/models.py +++ b/backend/python/app/models.py @@ -5,6 +5,7 @@ AIFlow 数据库模型定义 - SQLAlchemy ORM import uuid from datetime import datetime, date from sqlalchemy import ( + Text, Column, String, Integer, BigInteger, Float, Boolean, DateTime, Date, Text, Enum, ForeignKey, Index, JSON, DECIMAL, UniqueConstraint @@ -299,7 +300,7 @@ class MarketApp(Base): platform = Column(String(20), default=None, comment="Agent 接入平台：dify/n8n/coze") capability = Column(String(20), default=None, comment="能力分类：text/thinking/vision/tool/image/audio/asr/ocr/embedding/rerank/video") agent_endpoint = Column(String(500), default=None) - agent_api_key = Column(String(500), default=None, comment="单个Agent接入的API Key") + agent_api_key = Column(Text, default=None, comment="Agent 接入 API Key(3.2 起 Fernet 加密存储)") tool_policy = Column(JSON, default=None, comment="Agent 工具/MCP 权限沙箱策略(3.1)") created_at = Column(DateTime, default=datetime.now) updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now) diff --git a/migrations/023_widen_agent_api_key.sql b/migrations/023_widen_agent_api_key.sql new file mode 100644 index 0000000..6536507 --- /dev/null +++ b/migrations/023_widen_agent_api_key.sql @@ -0,0 +1,11 @@ +-- Phase 3.2: 放宽 market_apps.agent_api_key 容量以容纳 Fernet 密文 +SET @db = DATABASE(); +SET @tbl = 'market_apps'; +SET @col = 'agent_api_key'; +SELECT COLUMN_TYPE INTO @ctype FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=@db AND TABLE_NAME=@tbl AND COLUMN_NAME=@col; +SET @sql = IF(@ctype IS NULL OR @ctype NOT LIKE '%text%', + CONCAT('ALTER TABLE ', @tbl, ' MODIFY COLUMN ', @col, ' TEXT NULL COMMENT \'Agent 接入 API Key(3.2 起 Fernet 加密存储)\''), + 'SELECT 1'); +PREPARE stmt FROM @sql; +EXECUTE stmt; +DEALLOCATE PREPARE stmt;