(
app_name: str, dark_mode: bool = False, node: Optional[str] = None
)
| 514 | |
| 515 | @app.get("/dev/apps/{app_name}/build_graph_image") |
| 516 | async def get_app_info_image( |
| 517 | app_name: str, dark_mode: bool = False, node: Optional[str] = None |
| 518 | ) -> dict[str, GetEventGraphResult]: |
| 519 | runner = await self.get_runner_async(app_name) |
| 520 | |
| 521 | if not runner.app: |
| 522 | raise HTTPException( |
| 523 | status_code=404, detail=f"App not found: {app_name}" |
| 524 | ) |
| 525 | |
| 526 | app_info = serialize_app_info(runner.app) |
| 527 | |
| 528 | # Navigate to specific level if node is provided |
| 529 | if node: |
| 530 | target_agent = self._navigate_to_node(app_info, node) |
| 531 | if not target_agent: |
| 532 | raise HTTPException(status_code=404, detail=f"Node not found: {node}") |
| 533 | # Create a temporary app_info structure for the target level |
| 534 | app_info = {"root_agent": target_agent} |
| 535 | |
| 536 | workflows = self._get_all_sub_workflows(app_info, node if node else "") |
| 537 | |
| 538 | # This allows plotting non-workflow agents as a tree. |
| 539 | target_path = node if node else "" |
| 540 | if target_path not in workflows: |
| 541 | target_agent = app_info.get("root_agent") |
| 542 | if target_agent: |
| 543 | workflows[target_path] = target_agent |
| 544 | |
| 545 | results = {} |
| 546 | for path, info in workflows.items(): |
| 547 | dot_string = plot_workflow_graph( |
| 548 | {"root_agent": info}, format="dot", dark_mode=dark_mode |
| 549 | ) |
| 550 | if dot_string: |
| 551 | results[path] = GetEventGraphResult(dot_src=dot_string) |
| 552 | |
| 553 | return results |
| 554 | |
| 555 | # ========== AGENT TESTING ENDPOINTS ========== |
| 556 |
nothing calls this directly
no test coverage detected