(workflow, with_deep=False)
| 945 | |
| 946 | |
| 947 | def get_tool_id_list(workflow, with_deep=False): |
| 948 | from tools.models import ToolType, ToolWorkflow |
| 949 | |
| 950 | _result = [] |
| 951 | for node in workflow.get("nodes", []): |
| 952 | if node.get("type") == "tool-lib-node": |
| 953 | tool_id = node.get("properties", {}).get("node_data", {}).get("tool_lib_id") |
| 954 | if tool_id: |
| 955 | _result.append(tool_id) |
| 956 | elif node.get("type") == "loop-node": |
| 957 | r = get_tool_id_list(node.get("properties", {}).get("node_data", {}).get("loop_body", {})) |
| 958 | for item in r: |
| 959 | _result.append(item) |
| 960 | elif node.get("type") == "tool-workflow-lib-node": |
| 961 | tool_id = node.get("properties", {}).get("node_data", {}).get("tool_lib_id") |
| 962 | if tool_id: |
| 963 | _result.append(tool_id) |
| 964 | elif node.get("type") == "ai-chat-node": |
| 965 | node_data = node.get("properties", {}).get("node_data", {}) |
| 966 | mcp_tool_ids = node_data.get("mcp_tool_ids") or [] |
| 967 | skill_tool_ids = node_data.get("skill_tool_ids") or [] |
| 968 | tool_ids = node_data.get("tool_ids") or [] |
| 969 | for _id in mcp_tool_ids + tool_ids + skill_tool_ids: |
| 970 | _result.append(_id) |
| 971 | elif node.get("type") == "mcp-node": |
| 972 | mcp_tool_id = node.get("properties", {}).get("node_data", {}).get("mcp_tool_id") |
| 973 | if mcp_tool_id: |
| 974 | _result.append(mcp_tool_id) |
| 975 | if with_deep: |
| 976 | workflow_list = QuerySet(Tool).filter(id__in=_result, tool_type=ToolType.WORKFLOW) |
| 977 | tool_work_flow_list = QuerySet(ToolWorkflow).filter(tool_id__in=[wl.id for wl in workflow_list]) |
| 978 | for tool_work_flow in tool_work_flow_list: |
| 979 | child_tool_id_list = get_child_tool_id_list(tool_work_flow.work_flow, []) |
| 980 | for c in child_tool_id_list: |
| 981 | _result.append(c) |
| 982 | return _result |
| 983 | |
| 984 | |
| 985 | def get_child_tool_id_list(work_flow, response): |
no test coverage detected