Phase 3.3 多 Agent 编排图引擎: orchestration JSON 列 + 拓扑并行引擎 + local/external/reduce 节点 + 工具沙箱继承 + 后台编排 API

VERSION

@@ -1 +1 @@
-v2026.07.19.5
+v2026.07.19.8

backend/python/app/api/admin/agents_admin.py

@@ -88,6 +88,7 @@ def agent_to_dict(a: MarketApp) -> dict:
         "agent_endpoint": a.agent_endpoint,
         "skill_content": a.skill_content,
         "config": config,
+        "orchestration": a.orchestration,
         "created_at": a.created_at.isoformat() if a.created_at else None,
         "updated_at": a.updated_at.isoformat() if a.updated_at else None,
     }
@@ -555,3 +556,101 @@ async def get_available_models(
             for m, p in models
         ]
     }
+
+
+# ============================================================
+# 3.1 tool_policy 沙箱：Agent 工具/MCP 权限策略
+# ============================================================
+class ToolPolicyModel(BaseModel):
+    tool_policy: Optional[dict] = None  # null = 清除策略（恢复默认允许全部）
+
+
+@router.get("/{agent_id}/tool-policy", summary="获取 Agent 工具策略")
+async def get_agent_tool_policy(
+    agent_id: str,
+    admin_user=Depends(get_current_admin),
+    db: Session = Depends(get_db),
+):
+    agent = db.query(MarketApp).filter(
+        MarketApp.id == agent_id, MarketApp.type == "agent"
+    ).first()
+    if not agent:
+        raise HTTPException(status_code=404, detail="Agent 不存在")
+    return {"agent_id": agent_id, "tool_policy": agent.tool_policy}
+
+
+@router.put("/{agent_id}/tool-policy", summary="设置 Agent 工具策略")
+async def set_agent_tool_policy(
+    agent_id: str,
+    body: ToolPolicyModel,
+    admin_user=Depends(get_current_admin),
+    db: Session = Depends(get_db),
+):
+    agent = db.query(MarketApp).filter(
+        MarketApp.id == agent_id, MarketApp.type == "agent"
+    ).first()
+    if not agent:
+        raise HTTPException(status_code=404, detail="Agent 不存在")
+    tp = body.tool_policy
+    if tp is not None:
+        if not isinstance(tp, dict):
+            raise HTTPException(status_code=400, detail="tool_policy 必须为对象")
+        # 轻量结构校验
+        if "allowed_tools" in tp and tp["allowed_tools"] is not None:
+            at = tp["allowed_tools"]
+            if not isinstance(at, dict):
+                raise HTTPException(status_code=400, detail="allowed_tools 必须为对象")
+            for k in ("skill_ids", "mcp_ids"):
+                if k in at and at[k] is not None and not isinstance(at[k], list):
+                    raise HTTPException(status_code=400, detail=f"allowed_tools.{k} 必须为数组")
+        for k in ("max_calls_per_conversation", "per_round_cap"):
+            if k in tp and tp[k] is not None and not isinstance(tp[k], int):
+                raise HTTPException(status_code=400, detail=f"{k} 必须为整数")
+    agent.tool_policy = tp
+    db.commit()
+    return {"agent_id": agent_id, "tool_policy": agent.tool_policy, "updated": True}
+
+# 3.3 多 Agent 编排图(图引擎)
+# ============================================================
+class OrchestrationModel(BaseModel):
+    orchestration: Optional[dict] = None  # null = 清除编排图
+
+
+@router.get("/{agent_id}/orchestration", summary="获取 Agent 编排图")
+async def get_agent_orchestration(
+    agent_id: str,
+    admin_user=Depends(get_current_admin),
+    db: Session = Depends(get_db),
+):
+    agent = db.query(MarketApp).filter(
+        MarketApp.id == agent_id, MarketApp.type == "agent"
+    ).first()
+    if not agent:
+        raise HTTPException(status_code=404, detail="Agent 不存在")
+    return {"agent_id": agent_id, "orchestration": agent.orchestration}
+
+
+@router.put("/{agent_id}/orchestration", summary="设置 Agent 编排图")
+async def set_agent_orchestration(
+    agent_id: str,
+    body: OrchestrationModel,
+    admin_user=Depends(get_current_admin),
+    db: Session = Depends(get_db),
+):
+    agent = db.query(MarketApp).filter(
+        MarketApp.id == agent_id, MarketApp.type == "agent"
+    ).first()
+    if not agent:
+        raise HTTPException(status_code=404, detail="Agent 不存在")
+    og = body.orchestration
+    if og is not None:
+        if not isinstance(og, dict):
+            raise HTTPException(status_code=400, detail="orchestration 必须为对象")
+        try:
+            from app.core.orchestrator import validate_graph
+            validate_graph(og)
+        except Exception as e:
+            raise HTTPException(status_code=400, detail=f"编排图非法: {str(e)}")
+    agent.orchestration = og
+    db.commit()
+    return {"agent_id": agent_id, "orchestration": agent.orchestration, "updated": True}

backend/python/app/api/chat_agents.py

@@ -1,6 +1,8 @@
 # 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
@@ -900,7 +902,13 @@ async def _call_external_agent(db, agent: dict, query: str, user_id: str,
 
 
 
-    api_key = agent.get("agent_api_key") or ""
+    _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))
 
 
 

backend/python/app/api/chat_service.py

@@ -524,6 +524,29 @@ async def _send_message_core(
 
     external_agent = next((a for a in external_agents if a.get("agent_endpoint")), None)
 
+    # ===== 3.3 多 Agent 编排(图引擎)快路 =====
+    _orchestrator = next((a for a in external_agents if a.get("orchestration")), None)
+    if _orchestrator:
+        try:
+            from app.core.orchestrator import run_graph
+            _answer, _trace = await run_graph(
+                graph=_orchestrator["orchestration"],
+                query=req.content + attachment_block,
+                db=db,
+                user_id=user_id,
+                conv_id=conv_id,
+                trace_id=get_trace_id(),
+            )
+            ai_content = _answer or "编排未产生内容。"
+            logger.info(
+                f"[orchestrator] 图执行完成 trace_id={get_trace_id()} nodes={len(_trace.get('nodes', []))}"
+            )
+        except Exception as e:
+            logger.warning(f"[orchestrator] 图执行失败: {e}")
+            ai_content = f"⚠️ 多 Agent 编排执行失败：{str(e)[:200]}"
+        external_handled = True
+        external_agent = None  # 阻止下方外部 Agent 分支重复执行
+
 
 
     if external_agent:
@@ -1083,7 +1106,11 @@ async def _send_message_core(
 
 
 
+                    # ===== 3.1 tool_policy 沙箱：聚合策略 + 初始化预算计数器 =====
+                    _agent_tool_policy = _resolve_agent_tool_policy(agents)
+                    _tool_call_count = 0
                     for round_num in range(max_tool_rounds):
+                        _round_tool_count = 0
 
 
 
@@ -1297,6 +1324,26 @@ async def _send_message_core(
 
 
                                     func_name = tc.get("function", {}).get("name", "")
+                                    # ===== 3.1 tool_policy 沙箱：逐工具校验（白名单 + 预算 + 每轮上限） =====
+                                    _mcp_info = mcp_tool_map.get(func_name) if func_name.startswith("mcp_tool_") else None
+                                    _policy_check = _check_tool_policy(func_name, _mcp_info, skills, _agent_tool_policy, _tool_call_count, _round_tool_count)
+                                    if not _policy_check["allowed"]:
+                                        _reason = _policy_check["reason"]
+                                        logger.warning(
+                                            f"[tool_policy] blocked tool call: agent={_agent_tool_policy.get('agent_id') if _agent_tool_policy else None} "
+                                            f"tool={func_name} reason={_reason} trace={get_trace_id()}"
+                                        )
+                                        tool_result = f"[工具调用已被策略拦截] {_policy_check.get('message', _reason)}"
+                                        messages.append({
+                                            "role": "tool",
+                                            "tool_call_id": tc.get("id", f"call_{round_num}"),
+                                            "content": tool_result,
+                                        })
+                                        _tool_call_count += 1
+                                        _round_tool_count += 1
+                                        continue
+                                    _tool_call_count += 1
+                                    _round_tool_count += 1
 
 
 
@@ -2009,3 +2056,58 @@ async def _send_message_core(
 
 
     }
+
+
+# ======================================================================
+# 3.1 tool_policy 沙箱辅助函数
+# ======================================================================
+def _resolve_agent_tool_policy(agents):
+    """从已解析的 agents 聚合工具策略。多 agent 时取首个带有非空策略者。"""
+    if not agents:
+        return None
+    for ag in agents:
+        tp = ag.get("tool_policy")
+        if tp and isinstance(tp, dict):
+            tp = dict(tp)
+            tp["agent_id"] = ag.get("id")
+            return tp
+    return None
+
+
+def _check_tool_policy(func_name, mcp_info, skills, policy, call_count, round_count):
+    """校验单个工具调用是否被策略允许。
+
+    返回 {"allowed": bool, "reason": str, "tool_id": str|None, "message": str}
+    policy 为 None/空 -> 允许（向后兼容）。
+    """
+    if not policy:
+        return {"allowed": True, "tool_id": None}
+    allowed = policy.get("allowed_tools") or {}
+    skill_ids = allowed.get("skill_ids") or []
+    mcp_ids = allowed.get("mcp_ids") or []
+    is_mcp = func_name.startswith("mcp_tool_")
+    tool_id = None
+    if is_mcp:
+        if mcp_info and isinstance(mcp_info, dict):
+            tool_id = mcp_info.get("server_id")
+        if mcp_ids and tool_id not in mcp_ids:
+            return {"allowed": False, "reason": "mcp_not_in_allowlist", "tool_id": tool_id,
+                    "message": "MCP 服务不在该智能体的允许列表中"}
+    else:
+        if skills:
+            _s = next((s for s in skills if s.get("func_name") == func_name), None)
+            if _s:
+                tool_id = _s.get("id")
+        if skill_ids and tool_id not in skill_ids:
+            return {"allowed": False, "reason": "skill_not_in_allowlist", "tool_id": tool_id,
+                    "message": "技能不在该智能体的允许列表中"}
+    budget = policy.get("max_calls_per_conversation")
+    if budget is not None and call_count >= int(budget):
+        return {"allowed": False, "reason": "budget_exceeded", "tool_id": tool_id,
+                "message": policy.get("fallback_message")
+                or "该智能体已达到工具调用上限，请开启新对话或调整智能体设置。"}
+    per_round = policy.get("per_round_cap")
+    if per_round is not None and round_count >= int(per_round):
+        return {"allowed": False, "reason": "per_round_cap", "tool_id": tool_id,
+                "message": "本轮工具调用已达上限"}
+    return {"allowed": True, "tool_id": tool_id}

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
@@ -513,6 +514,7 @@ def _resolve_market_agents(db: Session, agent_ids: list, user_id: str) -> list:
 
 
 
+            "tool_policy": ag.tool_policy,
             # 外部 Agent 接入(Dify/n8n/Coze)运行期远程调用所需字段：
 
 
@@ -529,7 +531,8 @@ 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,
+            "orchestration": ag.orchestration,
 
 
 

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(),

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:]

backend/python/app/core/orchestrator.py

@@ -0,0 +1,395 @@
+"""app/core/orchestrator.py — Phase 3.3 多 Agent 编排图引擎。
+
+设计目标：
+- 以轻量 DAG 描述多 Agent 协作：节点(llm/local/external/reduce) + 有向边(数据流)。
+- 拓扑分层后按层并行执行(asyncio.gather)，互相独立的节点真正并发。
+- 失败隔离：每个节点异常不影响其它分支；per-node on_error=halt|continue。
+- 节点执行复用既有原语(_build_system_prompt / _execute_skill_tool / _call_external_agent /
+  3.1 工具沙箱)，完全向后兼容——agent.orchestration 为 null 时 chat_service 走原逻辑。
+- 并发安全：每个节点用独立 SessionLocal 会话做 DB 读，避免共享请求会话跨协程竞态。
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+from typing import Any, Optional
+
+import httpx
+
+logger = logging.getLogger("aiflow.orchestrator")
+
+VALID_NODE_TYPES = ("llm", "local", "external", "reduce")
+
+
+class OrchestrationError(ValueError):
+    """图结构非法或执行被 halt。"""
+
+
+# ---------------------------------------------------------------------------
+# 校验 + 拓扑
+# ---------------------------------------------------------------------------
+def validate_graph(graph: Any) -> None:
+    if not isinstance(graph, dict):
+        raise OrchestrationError("orchestration 必须为对象")
+    nodes = graph.get("nodes")
+    edges = graph.get("edges")
+    if not isinstance(nodes, list) or not nodes:
+        raise OrchestrationError("nodes 必须为非空数组")
+    if not isinstance(edges, list):
+        raise OrchestrationError("edges 必须为数组")
+
+    ids = set()
+    for i, n in enumerate(nodes):
+        if not isinstance(n, dict):
+            raise OrchestrationError(f"节点 #{i} 必须为对象")
+        nid = n.get("id")
+        if not nid or not isinstance(nid, str):
+            raise OrchestrationError(f"节点 #{i} 缺少字符串 id")
+        if nid in ids:
+            raise OrchestrationError(f"节点 id 重复: {nid}")
+        ids.add(nid)
+        ntype = n.get("type")
+        if ntype not in VALID_NODE_TYPES:
+            raise OrchestrationError(
+                f"节点 {nid} 类型非法: {ntype} (允许 {', '.join(VALID_NODE_TYPES)})"
+            )
+        if ntype in ("local", "external") and not n.get("agent_id"):
+            raise OrchestrationError(f"节点 {nid} 类型 {ntype} 必须指定 agent_id")
+
+    for i, e in enumerate(edges):
+        if not isinstance(e, dict):
+            raise OrchestrationError(f"边 #{i} 必须为对象")
+        f = e.get("from")
+        t = e.get("to")
+        if f not in ids:
+            raise OrchestrationError(f"边 #{i} 的 from={f} 指向不存在的节点")
+        if t not in ids:
+            raise OrchestrationError(f"边 #{i} 的 to={t} 指向不存在的节点")
+        if f == t:
+            raise OrchestrationError(f"边 #{i} 存在自环: {f}")
+
+
+def topological_layers(graph: dict) -> list:
+    """Kahn 算法返回分层(每层内节点可并行)；含环则抛错。"""
+    nodes = graph["nodes"]
+    edges = graph.get("edges", [])
+    indeg = {n["id"]: 0 for n in nodes}
+    adj = {n["id"]: [] for n in nodes}
+    for e in edges:
+        f, t = e["from"], e["to"]
+        adj[f].append(t)
+        indeg[t] += 1
+    order = {n["id"]: i for i, n in enumerate(nodes)}
+    layers: list = []
+    remaining = set(indeg)
+    while remaining:
+        layer = sorted([nid for nid in remaining if indeg[nid] == 0], key=lambda x: order[x])
+        if not layer:
+            cycle = sorted(remaining, key=lambda x: order[x])
+            raise OrchestrationError(f"检测到环路，无法拓扑排序: {cycle}")
+        layers.append(layer)
+        for nid in layer:
+            remaining.discard(nid)
+            for nxt in adj[nid]:
+                indeg[nxt] -= 1
+    return layers
+
+
+def _node_by_id(graph: dict, nid: str) -> dict:
+    for n in graph["nodes"]:
+        if n["id"] == nid:
+            return n
+    return {}
+
+
+def _predecessors(graph: dict, nid: str) -> list:
+    return [e["from"] for e in graph.get("edges", []) if e["to"] == nid]
+
+
+def _successors(graph: dict, nid: str) -> list:
+    return [e["to"] for e in graph.get("edges", []) if e["from"] == nid]
+
+
+def _find_output_node(graph: dict) -> str:
+    if graph.get("output_node"):
+        return graph["output_node"]
+    sinks = [n["id"] for n in graph["nodes"] if not _successors(graph, n["id"])]
+    if len(sinks) == 1:
+        return sinks[0]
+    for s in sinks:
+        if _node_by_id(graph, s).get("type") == "reduce":
+            return s
+    if sinks:
+        return sinks[0]
+    return graph["nodes"][-1]["id"]
+
+
+def _assemble_node_input(query: str, upstream: str, is_entry: bool) -> str:
+    """入口节点直接吃用户原问题；下游节点综合原问题 + 上游输出。"""
+    if is_entry:
+        return query or ""
+    if upstream:
+        return f"原始用户问题:\n{query}\n\n上游节点输出:\n{upstream}"
+    return query or ""
+
+
+# ---------------------------------------------------------------------------
+# 模型解析 + LLM 调用原语
+# ---------------------------------------------------------------------------
+async def _resolve_model_and_endpoint(model_id, db):
+    from app.models import AIModel, ModelProvider
+    from ..capability_registry import resolve_auto_model
+
+    ai_model = None
+    if model_id and model_id != "auto":
+        ai_model = db.query(AIModel).filter(AIModel.id == model_id).first()
+    if not ai_model:
+        ai_model = resolve_auto_model("text", db)
+    if not ai_model:
+        raise OrchestrationError("无法解析到可用模型")
+    provider = db.query(ModelProvider).filter(ModelProvider.id == ai_model.provider_id).first()
+    api_key = ai_model.api_key or (provider.api_key if provider else None)
+    api_url = ai_model.endpoint
+    if not api_url and provider and provider.api_base_url:
+        api_url = f"{provider.api_base_url.rstrip('/')}/v1/chat/completions"
+    if not api_url or not api_key:
+        raise OrchestrationError(f"节点模型 {ai_model.name} 缺少 endpoint 或 api_key")
+    return ai_model, api_url, api_key
+
+
+async def _llm_complete(system_prompt: str, user_content: str, model_id, db, user_id) -> str:
+    from app.api.chat_utils import _proxy_for_url
+    from app.core.observability import span
+
+    ai_model, api_url, api_key = await _resolve_model_and_endpoint(model_id, db)
+    messages = [
+        {"role": "system", "content": system_prompt},
+        {"role": "user", "content": user_content},
+    ]
+    request_body = {"model": ai_model.name, "messages": messages, "max_tokens": 2048}
+    async with span("orchestrator.llm", model=ai_model.id):
+        async with httpx.AsyncClient(
+            timeout=ai_model.timeout or 120,
+            proxy=_proxy_for_url(api_url),
+            trust_env=False,
+        ) as client:
+            resp = await client.post(
+                api_url,
+                headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
+                json=request_body,
+            )
+        if resp.status_code != 200:
+            raise OrchestrationError(f"LLM 调用失败 HTTP {resp.status_code}: {resp.text[:200]}")
+        data = resp.json()
+        return data["choices"][0]["message"].get("content") or ""
+
+
+# ---------------------------------------------------------------------------
+# local 节点：复用单 Agent system prompt + 受限 tool round + 3.1 沙箱
+# ---------------------------------------------------------------------------
+async def _run_local_agent_turn(agent: dict, node_input: str, db, user_id, conv_id) -> dict:
+    from app.api.chat_skill import _resolve_market_skills, _execute_skill_tool
+    from app.api.chat_context import _build_system_prompt, _retrieve_kb_context
+    from app.api.chat_service import _resolve_agent_tool_policy, _check_tool_policy
+    from app.api.chat_utils import _proxy_for_url
+
+    skills = _resolve_market_skills(db, agent.get("skill_ids") or [], user_id)
+    kb_context = _retrieve_kb_context(db, agent.get("knowledge_base_ids") or [], node_input)
+    system_prompt, function_definitions = _build_system_prompt(
+        skills=skills,
+        agents=[agent],
+        kb_context=kb_context,
+        user_content=node_input,
+        auto_discovered=False,
+    )
+    ai_model, api_url, api_key = await _resolve_model_and_endpoint(agent.get("model_id"), db)
+    tool_policy = _resolve_agent_tool_policy([agent])
+
+    messages = [
+        {"role": "system", "content": system_prompt},
+        {"role": "user", "content": node_input},
+    ]
+    called_skills: list = []
+    tool_call_count = 0
+    max_tool_rounds = 5
+    last_content = ""
+    async with httpx.AsyncClient(
+        timeout=ai_model.timeout or 120,
+        proxy=_proxy_for_url(api_url),
+        trust_env=False,
+    ) as client:
+        for round_num in range(max_tool_rounds):
+            request_body = {"model": ai_model.name, "messages": messages, "max_tokens": 2048}
+            if function_definitions:
+                request_body["tools"] = [
+                    {"type": "function", "function": fd} for fd in function_definitions
+                ]
+                request_body["tool_choice"] = "auto"
+            resp = await client.post(
+                api_url,
+                headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
+                json=request_body,
+            )
+            if resp.status_code != 200:
+                raise OrchestrationError(f"local 节点 LLM 失败 HTTP {resp.status_code}: {resp.text[:200]}")
+            data = resp.json()
+            message = data["choices"][0]["message"]
+            tool_calls = message.get("tool_calls") or []
+            content = message.get("content") or ""
+            last_content = content
+            if not tool_calls or not function_definitions:
+                return {"text": content, "used_tools": called_skills}
+            messages.append({"role": "assistant", "content": content, "tool_calls": tool_calls})
+            round_tool_count = 0
+            for tc in tool_calls:
+                func_name = tc.get("function", {}).get("name", "")
+                check = _check_tool_policy(
+                    func_name, None, skills, tool_policy, tool_call_count, round_tool_count
+                )
+                if not check["allowed"]:
+                    messages.append({
+                        "role": "tool",
+                        "tool_call_id": tc.get("id", f"call_{round_num}"),
+                        "content": f"[工具调用已被策略拦截] {check.get('message', check.get('reason'))}",
+                    })
+                    tool_call_count += 1
+                    round_tool_count += 1
+                    continue
+                tool_call_count += 1
+                round_tool_count += 1
+                try:
+                    func_args = json.loads(tc.get("function", {}).get("arguments", "{}") or "{}")
+                except Exception:
+                    func_args = {}
+                res = _execute_skill_tool(db, func_name, func_args, skills, node_input)
+                called_skills.append(func_name)
+                messages.append({
+                    "role": "tool",
+                    "tool_call_id": tc.get("id", f"call_{round_num}"),
+                    "content": res,
+                })
+    return {"text": last_content, "used_tools": called_skills}
+
+
+# ---------------------------------------------------------------------------
+# 单节点执行（含独立 DB 会话，保证并发安全）
+# ---------------------------------------------------------------------------
+async def _run_node(node, graph, blackboard, query, user_id, conv_id, trace_id) -> dict:
+    from app.api.chat_skill import _resolve_market_agents
+    from app.database import SessionLocal
+
+    nid = node["id"]
+    ntype = node["type"]
+    preds = _predecessors(graph, nid)
+    pred_results = [blackboard[p] for p in preds]
+    is_entry = not preds
+
+    if pred_results:
+        parts = []
+        for b in pred_results:
+            label = f"节点 {b.get('node_id')}/{b.get('node_type')}"
+            if b.get("ok"):
+                parts.append(f"【{label}】\n{b.get('output', '')}")
+            else:
+                parts.append(f"【{label} 执行失败】\n错误: {b.get('error', '')}")
+        upstream = "\n\n".join(parts)
+    else:
+        upstream = ""
+
+    with SessionLocal() as db:
+        if ntype in ("llm", "reduce"):
+            if ntype == "llm":
+                sys_prompt = node.get("system_prompt") or (
+                    "你是一个处理节点，请根据上游输入完成指定的文本变换、摘要或路由判断，"
+                    "只输出处理后的结果，不要多余解释。"
+                )
+            else:
+                sys_prompt = node.get("system_prompt") or (
+                    "你是一个结果汇总器，请将多个子任务的输出综合成连贯、完整、面向用户的最终回答。"
+                )
+            uc = _assemble_node_input(query, upstream, is_entry)
+            text = await _llm_complete(sys_prompt, uc, node.get("model_id"), db, user_id)
+            return {"ok": True, "node_id": nid, "node_type": ntype, "output": text}
+
+        if ntype == "local":
+            agents = _resolve_market_agents(db, [node["agent_id"]], user_id)
+            if not agents:
+                raise OrchestrationError(f"local 节点 {nid} 引用的 agent_id={node['agent_id']} 不存在")
+            ni = _assemble_node_input(query, upstream, is_entry)
+            res = await _run_local_agent_turn(agents[0], ni, db, user_id, conv_id)
+            return {
+                "ok": True,
+                "node_id": nid,
+                "node_type": "local",
+                "output": res["text"],
+                "used_tools": res.get("used_tools"),
+            }
+
+        if ntype == "external":
+            from app.api.chat_agents import _call_external_agent
+
+            agents = _resolve_market_agents(db, [node["agent_id"]], user_id)
+            if not agents:
+                raise OrchestrationError(f"external 节点 {nid} 引用的 agent_id={node['agent_id']} 不存在")
+            ni = _assemble_node_input(query, upstream, is_entry)
+            answer, _ = await _call_external_agent(
+                db, agents[0], query=ni, user_id=user_id, ext_conversation_id=None, history=[]
+            )
+            return {"ok": True, "node_id": nid, "node_type": "external", "output": answer or ""}
+
+    raise OrchestrationError(f"未知节点类型 {ntype}")
+
+
+# ---------------------------------------------------------------------------
+# 入口
+# ---------------------------------------------------------------------------
+async def run_graph(graph, query, db, user_id, conv_id, trace_id=None):
+    """执行编排图，返回 (answer: str, trace: dict)。
+
+    - 拓扑分层后逐层 asyncio.gather 并行执行。
+    - 节点异常按 on_error=halt|continue 处理（默认 continue，失败时下游拿到错误作为输入）。
+    """
+    validate_graph(graph)
+    layers = topological_layers(graph)
+    blackboard: dict = {}
+    trace = {"nodes": [], "trace_id": trace_id, "layers": layers}
+
+    for layer in layers:
+        tasks = [
+            _run_node(_node_by_id(graph, nid), graph, blackboard, query, user_id, conv_id, trace_id)
+            for nid in layer
+        ]
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+        for nid, res in zip(layer, results):
+            node = _node_by_id(graph, nid)
+            if isinstance(res, Exception):
+                on_error = node.get("on_error", "continue")
+                err_msg = str(res)
+                blackboard[nid] = {
+                    "ok": False,
+                    "node_id": nid,
+                    "node_type": node["type"],
+                    "error": err_msg,
+                }
+                trace["nodes"].append({"id": nid, "type": node["type"], "ok": False, "error": err_msg})
+                logger.warning(f"[orchestrator] 节点 {nid} 失败 on_error={on_error}: {err_msg}")
+                if on_error == "halt":
+                    raise OrchestrationError(f"节点 {nid} 执行失败(halt): {err_msg}")
+            else:
+                blackboard[nid] = res
+                trace["nodes"].append({
+                    "id": nid,
+                    "type": res.get("node_type"),
+                    "ok": True,
+                    "used_tools": res.get("used_tools"),
+                })
+
+    output_node = _find_output_node(graph)
+    final = blackboard.get(output_node, {})
+    if final.get("ok"):
+        answer = final.get("output", "")
+    else:
+        answer = f"⚠️ 编排输出节点 {output_node} 执行失败: {final.get('error', '未知错误')}"
+    trace["output_node"] = output_node
+    return answer, trace

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,9 @@ 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)")
+    orchestration = Column(JSON, default=None, comment="Agent 多Agent编排图(3.3)")
     created_at = Column(DateTime, default=datetime.now)
     updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
 

frontend/src/api/index.ts

@@ -887,6 +887,10 @@ export const adminAPI = {
 
   getAdminAgentReviews: (id: string) => get<any>(`/admin/agents/${id}/reviews`),
 
+  // Phase 3.1: 工具/MCP 权限沙箱 (tool_policy)
+  getAgentToolPolicy: (id: string) => get<any>(`/admin/agents/${id}/tool-policy`),
+  setAgentToolPolicy: (id: string, toolPolicy: any) => put<any>(`/admin/agents/${id}/tool-policy`, { tool_policy: toolPolicy }),
+
   getAdminAvailableSkills: (params?: any) => {
 
     const query = params ? '?' + Object.entries(params).filter(([_, v]) => v !== undefined && v !== '').map(([k, v]) => `${k}=${v}`).join('&') : '';

frontend/src/pages/admin/Agents.tsx

@@ -105,6 +105,10 @@ const Agents: React.FC = () => {
   // Available skills for linking
   const [availableSkills, setAvailableSkills] = useState<any[]>([]);
   const [availableModels, setAvailableModels] = useState<AvailableModel[]>([]);
+  // Available MCP servers for tool_policy sandbox allowlist
+  const [availableMcps, setAvailableMcps] = useState<any[]>([]);
+  // Tool policy (sandbox) — null means disabled / no policy
+  const [toolPolicy, setToolPolicy] = useState<any>(null);
 
   // Toast
   const [toast, setToast] = useState<{ msg: string; type: 'success' | 'error' } | null>(null);
@@ -144,14 +148,17 @@ const Agents: React.FC = () => {
   // Load available skills & models
   const loadAvailable = async () => {
     try {
-      const [skillsResult, modelsResult] = await Promise.all([
+      const [skillsResult, modelsResult, mcpsResult] = await Promise.all([
         adminAPI.getAdminAvailableSkills({ page_size: 50 }) as any,
         adminAPI.getAdminAvailableModels() as any,
+        adminAPI.getAdminMcpServers({ page_size: 50 }) as any,
       ]);
       const skills = (skillsResult.data || skillsResult).items || [];
       const models = (modelsResult.data || modelsResult) || [];
+      const mcps = (mcpsResult.data || mcpsResult).items || [];
       setAvailableSkills(skills);
       setAvailableModels(models);
+      setAvailableMcps(mcps);
     } catch {}
   };
 
@@ -187,6 +194,7 @@ const Agents: React.FC = () => {
       memory_enabled: false,
       reasoning_enabled: false,
     });
+    setToolPolicy(null);
     setShowModal(true);
   };
 
@@ -194,6 +202,7 @@ const Agents: React.FC = () => {
   const openEdit = async (agent: Agent) => {
     setEditingAgent(agent);
     await loadAvailable();
+    await openEditLoadPolicy(agent.id);
 
     const cfg = agent.config || {};
     setForm({
@@ -213,7 +222,15 @@ const Agents: React.FC = () => {
     setShowModal(true);
   };
 
-  // Open detail
+  // Fetch existing tool_policy (sandbox) for this agent
+  const openEditLoadPolicy = async (agentId: string) => {
+    try {
+      const tpRes = await adminAPI.getAgentToolPolicy(agentId) as any;
+      setToolPolicy(tpRes?.data?.tool_policy || null);
+    } catch {
+      setToolPolicy(null);
+    }
+  };
   const openDetail = async (agent: Agent) => {
     setDetailAgent(agent);
     setShowDetail(true);
@@ -246,7 +263,13 @@ const Agents: React.FC = () => {
     try {
       if (editingAgent) {
         await adminAPI.updateAdminAgent(editingAgent.id, payload);
-        showToast('Agent 已更新');
+        // Phase 3.1: persist tool_policy sandbox (null clears it)
+        try {
+          await adminAPI.setAgentToolPolicy(editingAgent.id, toolPolicy);
+          showToast('Agent 已更新');
+        } catch (pe: any) {
+          showToast(`Agent 已更新，但策略保存失败: ${pe.message || pe}`, 'error');
+        }
       } else {
         await adminAPI.createAdminAgent(payload);
         showToast('Agent 创建成功');
@@ -612,6 +635,87 @@ const Agents: React.FC = () => {
                   </div>
                 )}
               </div>
+              {editingAgent && (
+                <div style={{ gridColumn: '1 / -1' }}>
+                  <hr style={{ border: 'none', borderTop: '1px solid #E5E7EB', margin: '16px 0' }} />
+                  <div style={{ fontWeight: 600, marginBottom: 12 }}>🛡️ 工具权限沙箱 (tool_policy)</div>
+                  <label style={{ display: 'flex', alignItems: 'center', gap: 8, cursor: 'pointer', marginBottom: 12 }}>
+                    <input type="checkbox" checked={toolPolicy !== null} onChange={e => {
+                      if (e.target.checked) {
+                        setToolPolicy({ allowed_tools: { skill_ids: [], mcp_ids: [] }, max_calls_per_conversation: null, per_round_cap: null, fallback_message: '' });
+                      } else {
+                        setToolPolicy(null);
+                      }
+                    }} />
+                    启用工具/MCP 权限沙箱（限制该 Agent 可调用的工具与调用频次）
+                  </label>
+                  {toolPolicy && (
+                    <div style={{ display: 'flex', flexDirection: 'column', gap: 16 }}>
+                      <div className="form-group" style={{ marginBottom: 0 }}>
+                        <label className="form-label">允许调用的技能（不选 = 禁止调用任何技能）</label>
+                        <div style={{ display: 'flex', flexWrap: 'wrap', gap: 6, maxHeight: 160, overflowY: 'auto', padding: 8, border: '1px solid #E5E7EB', borderRadius: 6 }}>
+                          {availableSkills.length === 0 ? (
+                            <span style={{ color: '#9CA3AF', fontSize: 13 }}>暂无技能</span>
+                          ) : availableSkills.map((skill: any) => {
+                            const checked = (toolPolicy.allowed_tools?.skill_ids || []).includes(skill.id);
+                            return (
+                              <label key={skill.id} style={{ display: 'flex', alignItems: 'center', gap: 4, padding: '4px 10px', borderRadius: 12, fontSize: 12, cursor: 'pointer', background: checked ? '#EEF2FF' : '#F9FAFB', border: `1px solid ${checked ? '#6366F1' : '#E5E7EB'}`, color: checked ? '#6366F1' : '#374151' }}>
+                                <input type="checkbox" style={{ display: 'none' }} checked={checked} onChange={() => {
+                                  const cur = toolPolicy.allowed_tools?.skill_ids || [];
+                                  const next = checked ? cur.filter((x: string) => x !== skill.id) : [...cur, skill.id];
+                                  setToolPolicy({ ...toolPolicy, allowed_tools: { ...(toolPolicy.allowed_tools || {}), skill_ids: next } });
+                                }} />
+                                {skill.skill_name}
+                              </label>
+                            );
+                          })}
+                        </div>
+                      </div>
+                      <div className="form-group" style={{ marginBottom: 0 }}>
+                        <label className="form-label">允许调用的 MCP 服务（不选 = 禁止调用任何 MCP）</label>
+                        <div style={{ display: 'flex', flexWrap: 'wrap', gap: 6, maxHeight: 160, overflowY: 'auto', padding: 8, border: '1px solid #E5E7EB', borderRadius: 6 }}>
+                          {availableMcps.length === 0 ? (
+                            <span style={{ color: '#9CA3AF', fontSize: 13 }}>暂无 MCP 服务</span>
+                          ) : availableMcps.map((m: any) => {
+                            const checked = (toolPolicy.allowed_tools?.mcp_ids || []).includes(m.id);
+                            return (
+                              <label key={m.id} style={{ display: 'flex', alignItems: 'center', gap: 4, padding: '4px 10px', borderRadius: 12, fontSize: 12, cursor: 'pointer', background: checked ? '#EEF2FF' : '#F9FAFB', border: `1px solid ${checked ? '#6366F1' : '#E5E7EB'}`, color: checked ? '#6366F1' : '#374151' }}>
+                                <input type="checkbox" style={{ display: 'none' }} checked={checked} onChange={() => {
+                                  const cur = toolPolicy.allowed_tools?.mcp_ids || [];
+                                  const next = checked ? cur.filter((x: string) => x !== m.id) : [...cur, m.id];
+                                  setToolPolicy({ ...toolPolicy, allowed_tools: { ...(toolPolicy.allowed_tools || {}), mcp_ids: next } });
+                                }} />
+                                {m.name}
+                              </label>
+                            );
+                          })}
+                        </div>
+                      </div>
+                      <div style={{ display: 'grid', gridTemplateColumns: '1fr 1fr', gap: 16 }}>
+                        <div className="form-group" style={{ marginBottom: 0 }}>
+                          <label className="form-label">单对话最大调用次数（留空=不限制）</label>
+                          <input className="admin-form-input" type="number" min="0"
+                            value={toolPolicy.max_calls_per_conversation ?? ''}
+                            onChange={e => setToolPolicy({ ...toolPolicy, max_calls_per_conversation: e.target.value === '' ? null : parseInt(e.target.value) })} />
+                        </div>
+                        <div className="form-group" style={{ marginBottom: 0 }}>
+                          <label className="form-label">单轮最大调用次数（留空=不限制）</label>
+                          <input className="admin-form-input" type="number" min="0"
+                            value={toolPolicy.per_round_cap ?? ''}
+                            onChange={e => setToolPolicy({ ...toolPolicy, per_round_cap: e.target.value === '' ? null : parseInt(e.target.value) })} />
+                        </div>
+                      </div>
+                      <div className="form-group" style={{ marginBottom: 0 }}>
+                        <label className="form-label">额度耗尽提示语（可选）</label>
+                        <textarea className="admin-form-input" rows={2}
+                          value={toolPolicy.fallback_message || ''}
+                          onChange={e => setToolPolicy({ ...toolPolicy, fallback_message: e.target.value })}
+                          placeholder="例如：该智能体已达到工具调用上限，请开启新对话或调整智能体设置。" />
+                      </div>
+                    </div>
+                  )}
+                </div>
+              )}
             </div>
             <div className="modal-footer">
               <button className="admin-btn" onClick={() => setShowModal(false)}>取消</button>

migrations/022_add_tool_policy.sql

@@ -0,0 +1,15 @@
+-- Phase 3.1: MarketApp 增加 tool_policy（Agent 工具/MCP 权限沙箱策略）
+-- 幂等：仅当列不存在时才 ALTER
+SET @db = 'aiflow';
+SET @tbl = 'market_apps';
+SET @col = 'tool_policy';
+SET @exists = (
+    SELECT COUNT(*) FROM information_schema.columns
+    WHERE table_schema = @db AND table_name = @tbl AND column_name = @col
+);
+SET @sql = IF(@exists = 0,
+    'ALTER TABLE market_apps ADD COLUMN tool_policy JSON NULL',
+    'SELECT 1');
+PREPARE stmt FROM @sql;
+EXECUTE stmt;
+DEALLOCATE PREPARE stmt;

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;

migrations/024_add_orchestration.sql

@@ -0,0 +1,15 @@
+-- Phase 3.3: MarketApp 增加 orchestration（Agent 多 Agent 编排图，JSON）
+-- 幂等：仅当列不存在时才 ALTER
+SET @db = 'aiflow';
+SET @tbl = 'market_apps';
+SET @col = 'orchestration';
+SET @exists = (
+    SELECT COUNT(*) FROM information_schema.columns
+    WHERE table_schema = @db AND table_name = @tbl AND column_name = @col
+);
+SET @sql = IF(@exists = 0,
+    "ALTER TABLE market_apps ADD COLUMN orchestration JSON NULL COMMENT 'Agent 多Agent编排图(3.3)'",
+    'SELECT 1');
+PREPARE stmt FROM @sql;
+EXECUTE stmt;
+DEALLOCATE PREPARE stmt;