Coverage for src / mcp_server_langgraph / builder / importer / importer.py: 100%
31 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-12-03 00:43 +0000
1"""
2Main Importer Module
4High-level API for importing Python code into visual builder.
6Combines:
7- AST parsing
8- Graph extraction
9- Auto-layout
10- Type inference
12Example:
13 from mcp_server_langgraph.builder.importer import import_from_file
15 # Import existing agent code
16 workflow = import_from_file("src/agents/research_agent.py")
18 # Use in visual builder
19 # - workflow["nodes"] has positions for canvas
20 # - workflow["edges"] ready for React Flow
21 # - workflow["state_schema"] for configuration
23 # Or import code string
24 workflow = import_from_code(python_code_string)
25"""
27from typing import Any, Literal
29from .graph_extractor import GraphExtractor
30from .layout_engine import LayoutEngine
33def import_from_code(code: str, layout_algorithm: Literal["hierarchical", "force", "grid"] = "hierarchical") -> dict[str, Any]:
34 """
35 Import workflow from Python code string.
37 Args:
38 code: Python source code
39 layout_algorithm: Layout algorithm for positioning
41 Returns:
42 Complete workflow definition ready for visual builder
44 Example:
45 >>> workflow = import_from_code(python_code)
46 >>> len(workflow["nodes"])
47 5
48 >>> workflow["nodes"][0]["position"]
49 {'x': 250, 'y': 50}
50 """
51 # Extract workflow structure
52 extractor = GraphExtractor()
53 workflow = extractor.extract_from_code(code)
55 # Auto-layout nodes
56 layout_engine = LayoutEngine()
57 positioned_nodes = layout_engine.layout(
58 workflow["nodes"], workflow["edges"], algorithm=layout_algorithm, entry_point=workflow.get("entry_point")
59 )
61 # Update workflow with positioned nodes
62 workflow["nodes"] = positioned_nodes
64 return workflow
67def import_from_file(
68 file_path: str, layout_algorithm: Literal["hierarchical", "force", "grid"] = "hierarchical"
69) -> dict[str, Any]:
70 """
71 Import workflow from Python file.
73 Args:
74 file_path: Path to Python file
75 layout_algorithm: Layout algorithm
77 Returns:
78 Workflow definition
80 Example:
81 >>> workflow = import_from_file("src/agents/my_agent.py")
82 >>> # Load into visual builder
83 >>> builder.load_workflow(workflow)
84 """
85 with open(file_path) as f:
86 code = f.read()
88 return import_from_code(code, layout_algorithm)
91def validate_import(workflow: dict[str, Any]) -> dict[str, Any]:
92 """
93 Validate imported workflow.
95 Args:
96 workflow: Imported workflow
98 Returns:
99 Validation results
101 Example:
102 >>> workflow = import_from_file("agent.py")
103 >>> validation = validate_import(workflow)
104 >>> validation["valid"]
105 True
106 """
107 errors = []
108 warnings = []
110 # Check required fields
111 required = ["name", "nodes", "edges"]
112 for field in required:
113 if field not in workflow:
114 errors.append(f"Missing required field: {field}")
116 # Check nodes have positions
117 for node in workflow.get("nodes", []):
118 if "position" not in node:
119 errors.append(f"Node {node['id']} missing position")
121 # Warn about empty state schema
122 if not workflow.get("state_schema"):
123 warnings.append("No state schema extracted - may need manual configuration")
125 # Warn about unknown node types
126 known_types = ["tool", "llm", "conditional", "approval", "custom"]
127 for node in workflow.get("nodes", []):
128 if node.get("type") not in known_types:
129 warnings.append(f"Node {node['id']} has unknown type: {node.get('type')}")
131 return {"valid": len(errors) == 0, "errors": errors, "warnings": warnings}
134# ==============================================================================
135# Example Usage
136# ==============================================================================
138if __name__ == "__main__":
139 # Sample LangGraph code
140 sample_code = '''
141from typing import TypedDict, List
142from langgraph.graph import StateGraph
145class ResearchState(TypedDict):
146 """Research agent state."""
147 query: str
148 search_results: List[str]
149 summary: str
150 confidence: float
153def search_web(state):
154 """Search web for information."""
155 return state
158def filter_results(state):
159 """Filter search results."""
160 return state
163def summarize(state):
164 """Summarize findings."""
165 return state
168def create_research_agent():
169 """Create research agent."""
170 graph = StateGraph(ResearchState)
172 graph.add_node("search", search_web)
173 graph.add_node("filter", filter_results)
174 graph.add_node("summarize", summarize)
176 graph.add_edge("search", "filter")
177 graph.add_edge("filter", "summarize")
179 graph.set_entry_point("search")
180 graph.set_finish_point("summarize")
182 return graph.compile()
183'''
185 print("=" * 80)
186 print("CODE IMPORTER - TEST RUN")
187 print("=" * 80)
189 # Import and layout
190 workflow = import_from_code(sample_code, layout_algorithm="hierarchical")
192 print(f"\nWorkflow: {workflow['name']}")
193 print(f"Nodes: {len(workflow['nodes'])}")
194 print(f"Edges: {len(workflow['edges'])}")
196 print("\nNode Positions (Hierarchical Layout):")
197 for node in workflow["nodes"]:
198 print(f" {node['id']:15} → ({node['position']['x']:6.1f}, {node['position']['y']:6.1f}) [{node['type']}]")
200 # Validate
201 validation = validate_import(workflow)
202 print(f"\nValidation: {'✅ PASSED' if validation['valid'] else '❌ FAILED'}")
203 if validation["warnings"]:
204 print("Warnings:")
205 for warning in validation["warnings"]:
206 print(f" ⚠️ {warning}")
208 # Try different layouts
209 print("\n" + "=" * 80)
210 print("LAYOUT COMPARISON")
211 print("=" * 80)
213 for layout_alg in ["hierarchical", "force", "grid"]:
214 workflow_variant = import_from_code(sample_code, layout_algorithm=layout_alg) # type: ignore
215 print(f"\n{layout_alg.upper()} layout:")
216 for node in workflow_variant["nodes"][:3]: # Show first 3
217 print(f" {node['id']:15} @ ({node['position']['x']:6.1f}, {node['position']['y']:6.1f})")
219 print("\n" + "=" * 80)