Expose Bisheng workflows to external agents via MCP / 通过 MCP 向外部 Agent 开放 Bisheng 工作流能力

Expose Bisheng workflow capabilities to external agents through MCP, so outside agents can discover, read, author, and invoke workflows through a standardized protocol instead of relying only on the Bisheng product UI. This change also adds the key pieces needed for real usage, including workflow authoring, condition-node compatibility, authentication, and device-flow login, making external integration both practical and controlled.
通过 MCP 向外部 Agent 开放 Bisheng 工作流能力，使外部智能体可以通过标准化协议发现、读取、编排并调用工作流，而不再只能依赖 Bisheng 站内页面完成操作。此次改动同时补齐了工作流 authoring、条件节点兼容、鉴权与 device flow 登录等关键能力，让外部接入在可用性和安全性上都达到可落地状态。

Constraint: Need a clean commit boundary before switching to upstream/2.5.0-PM / 切换到 upstream/2.5.0-PM 前需要一个清晰的提交边界
Rejected: git stash | less visible and less durable than a branch-local checkpoint commit / 拒绝使用 git stash：可见性和持久性都不如分支内检查点提交
Confidence: medium / 中
Scope-risk: narrow / 范围窄
Reversibility: clean / 可干净回退
Directive: Curate or squash this checkpoint before merge if a cleaner history is required / 如需更干净的历史，请在合并前整理或压缩这个检查点提交
Tested: src/backend/.venv/bin/python -m pytest -q src/backend/test/test_mcp_device_flow.py src/backend/test/test_external_workflow_service.py; src/backend/.venv/bin/python -m pytest -q src/backend/test/test_workflow_mcp_e2e.py / 已验证：上述 MCP 相关后端测试
Not-tested: Full frontend build and manual end-to-end environment verification / 未验证：完整前端构建与手工端到端环境联调

Author: Y1fe1Zh0u

.gitignore

@@ -264,3 +264,4 @@ docker/office/bisheng/*.gz
 
 CLAUDE.md
 !src/backend/bisheng/telemetry_search/**/*.pyc
+.omx/

AGENT.md

@@ -0,0 +1,89 @@
+## 基本行为约束
+
+### 1. 直接解决问题，禁止废话
+- 不要寒暄
+- 不要重复用户问题
+- 不要输出无信息量的过渡句、总结句、鼓励句、评价句
+- 不要使用“可以”“当然”“没问题”“你这个问题很好”之类的表达
+- 直接进入分析、判断和方案输出
+
+### 2. 优先永久性解决方案，禁止临时绕过
+- 默认优先选择可维护、可复用、可扩展、可验证的方案
+- 如果存在“根因修复”和“临时绕过”两种路径，优先输出根因修复
+- 不要把 hack、patch、手工补丁、一次性操作伪装成正式方案
+- 如果只能提供临时方案，必须明确标注这是临时措施，并说明其局限、风险、替代的正式方案
+
+### 3. 结合上下文理解用户目的，禁止直接猜测
+- 回答前先判断：
+  - 用户显式提出的问题是什么
+  - 用户真正要达成的目标是什么
+  - 当前上下文中有哪些已知约束
+- 不要脱离上下文按字面机械作答
+- 不要凭空补充不存在的前提
+- 当信息不足时：
+  - 先基于已有信息给出条件化判断
+  - 明确指出哪些部分是已知，哪些部分未知
+  - 不要把推测当成事实
+
+## 决策原则
+
+### 根因优先
+输出方案时，优先按以下顺序思考：
+1. 问题的根因是什么
+2. 是否可以从结构、流程、接口、数据、权限、配置、架构层面彻底解决
+3. 该方案是否能避免同类问题再次发生
+4. 该方案的维护成本是否可接受
+
+### 用户目标优先
+不要只回答表面问题，要判断用户是在：
+- 要一个定义
+- 要一个判断
+- 要一个方案
+- 要一个可落地实现
+- 要一个权衡分析
+输出必须匹配用户真实任务层级。
+
+### 上下文一致性
+如果用户前文已经给出：
+- 业务背景
+- 系统架构
+- 术语定义
+- 限制条件
+- 偏好方案
+则必须沿用这些上下文，不要重新发明一套设定。
+
+## 输出要求
+
+### 信息组织
+优先输出以下结构之一，按任务类型自动选择：
+- 问题分析 → 根因 → 方案 → 风险/边界
+- 目标 → 约束 → 可选方案对比 → 推荐方案
+- 结论 → 依据 → 落地步骤
+- 定义 → 与相近概念的区别 → 在当前场景下的含义
+
+### 表达要求
+- 用精确术语，不用空泛表达
+- 能具体就具体，不要泛泛而谈
+- 能落地到机制、规则、数据结构、流程，就不要只停留在概念层
+- 不要为了显得全面而输出无关内容
+
+## 禁止行为
+- 禁止为了“显得聪明”而过度延展
+- 禁止把不确定内容说成确定事实
+- 禁止忽略用户已有上下文重新回答
+- 禁止优先给表面 workaround 而不说明正式方案
+- 禁止只给正确但不可执行的抽象建议
+
+## 遇到信息不足时
+不要直接猜答案。改为：
+1. 先明确当前已知信息
+2. 给出在这些信息下最合理的分析
+3. 标出需要额外信息才可确定的部分
+4. 如果可能，给出分条件方案
+
+## 最终标准
+你的回答必须同时满足：
+- 对准用户真正目标
+- 尽量一次解决，而不是临时糊住
+- 与已有上下文一致
+- 结论可执行、可维护、可验证

docker/bisheng/entrypoint.sh

@@ -29,9 +29,14 @@ start_default(){
     celery -A bisheng.worker.main worker -l info -c 100 -P threads -Q celery -n celery@%h
 }
 
+start_min_worker(){
+    # 最小化worker进程数，减少资源占用
+    celery -A bisheng.worker.main worker -l info -c 100 -P threads -Q knowledge_celery,workflow_celery,celery -n min_worker@%h
+}
+
 if [ "$start_mode" = "api" ]; then
     echo "Starting API server..."
-    uvicorn bisheng.main:app --host 0.0.0.0 --port 7860 --no-access-log --workers 8
+    uvicorn bisheng.main:app --host 0.0.0.0 --port 7860 --no-access-log --workers 1
 elif [ "$start_mode" = "knowledge" ]; then
     echo "Starting Knowledge Celery worker..."
     start_knowledge
@@ -49,14 +54,8 @@ elif [ "$start_mode" = "linsight" ]; then
     start_linsight
 elif [ "$start_mode" = "worker" ]; then
     echo "Starting All worker..."
-    # 处理知识库相关任务的worker
-    start_knowledge &
-    # 处理工作流相关任务的worker
-    start_workflow &
-    # 处理linsight相关任务的worker
-    start_linsight &
-    # 默认其他任务的执行worker，目前是定时统计埋点数据
-    start_default &
+    # 最小化worker进程数，减少资源占用
+    start_min_worker &
     start_beat
 
     echo "All workers started successfully."

docker/docker-compose.override.yml

@@ -0,0 +1,8 @@
+services:
+  backend:
+    volumes:
+      - ../src/backend/bisheng:/app/bisheng
+
+  backend_worker:
+    volumes:
+      - ../src/backend/bisheng:/app/bisheng

src/backend/bisheng/api/services/external_workflow.py

@@ -761,6 +761,141 @@ def _ensure_create_graph_scaffold(cls, graph_data: dict) -> dict:
 
         return updated_graph
 
+    @classmethod
+    def _extract_descriptor_param_updates(cls, node_descriptor: dict) -> dict:
+        params = node_descriptor.get('params')
+        if not isinstance(params, dict):
+            return {}
+        updates = {}
+        for key, value in params.items():
+            if isinstance(value, dict) and 'value' in value:
+                updates[key] = copy.deepcopy(value.get('value'))
+            else:
+                updates[key] = copy.deepcopy(value)
+        return updates
+
+    @classmethod
+    def _resolve_descriptor_position(cls,
+                                     node_descriptor: dict,
+                                     existing_node: Optional[dict],
+                                     *,
+                                     axis: str,
+                                     fallback: float) -> float:
+        position = node_descriptor.get('position')
+        if isinstance(position, dict) and axis in position:
+            try:
+                return float(position.get(axis))
+            except Exception:
+                cls._raise_workflow_error(
+                    f'Workflow node {node_descriptor.get("id") or "unknown"} has an invalid {axis} position'
+                )
+        if existing_node is not None:
+            return cls._node_position_value(existing_node, axis)
+        return fallback
+
+    @classmethod
+    def _build_graph_node_from_descriptor(cls,
+                                          node_descriptor: dict,
+                                          *,
+                                          existing_node: Optional[dict] = None,
+                                          node_index: int = 0) -> dict:
+        node_id = node_descriptor.get('id') or (existing_node or {}).get('id')
+        if not node_id:
+            cls._raise_workflow_error('Workflow node descriptor must contain an id')
+
+        node_type = node_descriptor.get('type') or (cls._get_node_type(existing_node) if existing_node else '')
+        if not node_type:
+            cls._raise_workflow_error(f'Workflow node descriptor {node_id} must contain a type')
+
+        name = node_descriptor.get('name')
+        description = node_descriptor.get('description')
+        tab = node_descriptor.get('tab')
+        position_x = cls._resolve_descriptor_position(
+            node_descriptor,
+            existing_node,
+            axis='x',
+            fallback=float(node_index * cls._DEFAULT_HORIZONTAL_NODE_GAP),
+        )
+        position_y = cls._resolve_descriptor_position(
+            node_descriptor,
+            existing_node,
+            axis='y',
+            fallback=0.0,
+        )
+
+        if existing_node is not None and cls._get_node_type(existing_node) == node_type:
+            node_payload = copy.deepcopy(existing_node)
+        else:
+            node_payload = create_graph_node_payload(
+                node_type,
+                node_id=node_id,
+                name=name or '',
+                position_x=position_x,
+                position_y=position_y,
+            )
+            if node_payload is None:
+                raise NotFoundError(msg=f'Workflow node template not found: {node_type}')
+
+        node_payload['id'] = node_id
+        node_payload['position'] = {'x': position_x, 'y': position_y}
+        node_data = node_payload.setdefault('data', {})
+        node_data['id'] = node_id
+        node_data['type'] = node_type
+        if name is not None:
+            node_data['name'] = name
+        else:
+            node_data.setdefault('name', '')
+        if description is not None:
+            node_data['description'] = description
+        if tab is not None:
+            node_data['tab'] = copy.deepcopy(tab)
+
+        param_updates = cls._extract_descriptor_param_updates(node_descriptor)
+        if param_updates:
+            cls._patch_node_fields(node_payload, param_updates)
+
+        return node_payload
+
+    @classmethod
+    def _coerce_editor_graph_input(cls,
+                                   graph_data: dict,
+                                   *,
+                                   base_graph: Optional[dict] = None) -> dict:
+        if not isinstance(graph_data, dict):
+            return graph_data
+
+        nodes = graph_data.get('nodes')
+        if not isinstance(nodes, list):
+            return graph_data
+
+        if all(not isinstance(node, dict) or isinstance(node.get('data'), dict) for node in nodes):
+            return cls._normalize_editor_node_types(copy.deepcopy(graph_data))
+
+        base_nodes_by_id = {}
+        if isinstance(base_graph, dict):
+            for existing_node in base_graph.get('nodes', []):
+                if isinstance(existing_node, dict) and existing_node.get('id'):
+                    base_nodes_by_id[existing_node['id']] = existing_node
+
+        updated_graph = copy.deepcopy(graph_data)
+        rebuilt_nodes = []
+        for index, raw_node in enumerate(nodes):
+            if not isinstance(raw_node, dict):
+                rebuilt_nodes.append(raw_node)
+                continue
+            if isinstance(raw_node.get('data'), dict):
+                rebuilt_nodes.append(copy.deepcopy(raw_node))
+                continue
+            rebuilt_nodes.append(
+                cls._build_graph_node_from_descriptor(
+                    raw_node,
+                    existing_node=base_nodes_by_id.get(raw_node.get('id')),
+                    node_index=index,
+                )
+            )
+        updated_graph['nodes'] = rebuilt_nodes
+        return cls._normalize_editor_node_types(updated_graph)
+
     @classmethod
     def _apply_node_initial_params(cls, node_payload: dict, initial_params: Optional[dict]) -> dict:
         if not initial_params:
@@ -996,6 +1131,7 @@ def _create_workflow_draft_sync(cls,
                                     description: Optional[str] = None,
                                     guide_word: Optional[str] = None) -> tuple[Flow, FlowVersion]:
         cls._assert_workflow_name_available(login_user, name)
+        graph_data = cls._coerce_editor_graph_input(graph_data)
         graph_data = cls._ensure_create_graph_scaffold(graph_data)
         cls._validate_draft_graph(login_user, graph_data, flow_name=name)
 
@@ -1054,7 +1190,7 @@ async def update_workflow_draft(cls,
         if name is not None and name != flow.name:
             cls._assert_workflow_name_available(login_user, name, exclude_flow_id=flow.id)
 
-        graph_data = cls._normalize_editor_node_types(copy.deepcopy(graph_data))
+        graph_data = cls._coerce_editor_graph_input(graph_data, base_graph=editable_version.data)
         if has_flow_updates:
             if name is not None:
                 flow.name = name

src/backend/bisheng/mcp_server/auth.py

@@ -127,6 +127,10 @@ def _hash_session_token(token: str) -> str:
     return hashlib.sha256(token.encode('utf-8')).hexdigest()
 
 
+def hash_bisheng_session_token(token: str) -> str:
+    return _hash_session_token(token)
+
+
 def _normalize_scopes(scopes: Optional[list[str] | tuple[str, ...]]) -> tuple[str, ...]:
     if not scopes:
         return _MCP_DEFAULT_SCOPES
@@ -137,6 +141,12 @@ def _normalize_scopes(scopes: Optional[list[str] | tuple[str, ...]]) -> tuple[st
     return tuple(normalized or _MCP_DEFAULT_SCOPES)
 
 
+def normalize_mcp_scopes(scopes: Optional[list[str] | tuple[str, ...] | str]) -> tuple[str, ...]:
+    if isinstance(scopes, str):
+        scopes = [scope.strip() for scope in scopes.replace(',', ' ').split() if scope.strip()]
+    return _normalize_scopes(scopes)
+
+
 async def _assert_parent_session_valid(user_id: int, parent_session_hash: str):
     redis_client = await get_redis_client()
     current_session = await redis_client.aget(USER_CURRENT_SESSION.format(user_id))
@@ -155,11 +165,11 @@ async def resolve_login_user_from_bisheng_access_token(token: str) -> UserPayloa
     )
 
 
-def create_mcp_access_token(login_user: LoginUser,
-                            parent_access_token: str,
-                            *,
-                            scopes: Optional[list[str] | tuple[str, ...]] = None,
-                            expires_in: int = _MCP_DEFAULT_EXPIRES_IN) -> tuple[str, dict]:
+def _create_mcp_access_token_from_session_hash(login_user: LoginUser,
+                                               parent_session_hash: str,
+                                               *,
+                                               scopes: Optional[list[str] | tuple[str, ...]] = None,
+                                               expires_in: int = _MCP_DEFAULT_EXPIRES_IN) -> tuple[str, dict]:
     now = int(datetime.now(timezone.utc).timestamp())
     expires_in = max(60, min(int(expires_in), _MCP_MAX_EXPIRES_IN))
     normalized_scopes = list(_normalize_scopes(scopes))
@@ -174,7 +184,7 @@ def create_mcp_access_token(login_user: LoginUser,
         'jti': generate_uuid(),
         'token_type': _MCP_TOKEN_TYPE,
         'scope': normalized_scopes,
-        'parent_session_hash': _hash_session_token(parent_access_token),
+        'parent_session_hash': parent_session_hash,
     }
     token = jwt.encode(claims, AuthJwt().jwt_secret, algorithm='HS256')
     return token, {
@@ -186,6 +196,32 @@ def create_mcp_access_token(login_user: LoginUser,
     }
 
 
+def create_mcp_access_token(login_user: LoginUser,
+                            parent_access_token: str,
+                            *,
+                            scopes: Optional[list[str] | tuple[str, ...]] = None,
+                            expires_in: int = _MCP_DEFAULT_EXPIRES_IN) -> tuple[str, dict]:
+    return _create_mcp_access_token_from_session_hash(
+        login_user,
+        _hash_session_token(parent_access_token),
+        scopes=scopes,
+        expires_in=expires_in,
+    )
+
+
+def create_mcp_access_token_from_session_hash(login_user: LoginUser,
+                                              parent_session_hash: str,
+                                              *,
+                                              scopes: Optional[list[str] | tuple[str, ...]] = None,
+                                              expires_in: int = _MCP_DEFAULT_EXPIRES_IN) -> tuple[str, dict]:
+    return _create_mcp_access_token_from_session_hash(
+        login_user,
+        parent_session_hash,
+        scopes=scopes,
+        expires_in=expires_in,
+    )
+
+
 async def _validate_mcp_access_token(token: str) -> tuple[UserPayload, tuple[str, ...]]:
     try:
         payload = jwt.decode(