Navigate to a specific node in the agent hierarchy. Args: app_info: The full app info structure node_path: Path like "agent1/agent2/agent3" Returns: The agent data at that path, or None if not found
(self, app_info: dict, node_path: str)
| 1188 | return {} |
| 1189 | |
| 1190 | def _navigate_to_node(self, app_info: dict, node_path: str) -> dict | None: |
| 1191 | """Navigate to a specific node in the agent hierarchy. |
| 1192 | |
| 1193 | Args: |
| 1194 | app_info: The full app info structure |
| 1195 | node_path: Path like "agent1/agent2/agent3" |
| 1196 | |
| 1197 | Returns: |
| 1198 | The agent data at that path, or None if not found |
| 1199 | """ |
| 1200 | if not node_path: |
| 1201 | return app_info.get("root_agent") |
| 1202 | |
| 1203 | # Strip leading/trailing slashes and split, filter out empty strings |
| 1204 | path_parts = [p for p in node_path.strip("/").split("/") if p] |
| 1205 | current = app_info.get("root_agent") |
| 1206 | |
| 1207 | if not current: |
| 1208 | return None |
| 1209 | |
| 1210 | # Navigate through each level (skip first if it's the root name) |
| 1211 | start_idx = 0 |
| 1212 | if path_parts[0] == current.get("name"): |
| 1213 | start_idx = 1 |
| 1214 | |
| 1215 | for part in path_parts[start_idx:]: |
| 1216 | found = None |
| 1217 | # Check potential containers in order of preference |
| 1218 | containers = [] |
| 1219 | if current.get("graph") and current["graph"].get("nodes"): |
| 1220 | containers.append(current["graph"]["nodes"]) |
| 1221 | if current.get("nodes"): |
| 1222 | containers.append(current["nodes"]) |
| 1223 | if current.get("sub_agents"): |
| 1224 | containers.append(current["sub_agents"]) |
| 1225 | |
| 1226 | for container in containers: |
| 1227 | for item in container: |
| 1228 | if item.get("name") == part: |
| 1229 | found = item |
| 1230 | break |
| 1231 | if found: |
| 1232 | break |
| 1233 | |
| 1234 | if not found: |
| 1235 | return None |
| 1236 | current = found |
| 1237 | |
| 1238 | return current |
| 1239 | |
| 1240 | def _get_all_sub_workflows( |
| 1241 | self, app_info: dict, current_path: str = "" |
no test coverage detected