Coverage for src/quickmcp/factory_old.py: 0%
374 statements
« prev ^ index » next coverage.py v7.10.4, created at 2025-08-20 17:51 +0200
1"""
2QuickMCP Factory - Automatically generate MCP servers from Python code
3"""
5import ast
6import inspect
7import importlib.util
8import sys
9from pathlib import Path
10from typing import Any, Dict, List, Optional, Callable, get_type_hints, Set
11import json
12import logging
13import re
14import traceback
15from dataclasses import dataclass
17from .server import QuickMCPServer
19logger = logging.getLogger(__name__)
22@dataclass
23class MissingDependency:
24 """Information about a missing dependency."""
25 module: str
26 import_type: str # "import", "from_import", "optional"
27 line_number: Optional[int] = None
28 source_line: Optional[str] = None
29 suggested_install: Optional[str] = None
32class ImportAnalyzer:
33 """Analyze Python files for imports and missing dependencies."""
35 # Common package name to pip install name mappings
36 PIP_MAPPINGS = {
37 'aiohttp': 'aiohttp',
38 'requests': 'requests',
39 'fastapi': 'fastapi',
40 'starlette': 'starlette',
41 'uvicorn': 'uvicorn',
42 'pydantic': 'pydantic',
43 'numpy': 'numpy',
44 'pandas': 'pandas',
45 'sqlalchemy': 'sqlalchemy',
46 'psycopg2': 'psycopg2-binary',
47 'mysql': 'mysql-connector-python',
48 'redis': 'redis',
49 'celery': 'celery',
50 'pytest': 'pytest',
51 'click': 'click',
52 'typer': 'typer',
53 'rich': 'rich',
54 'httpx': 'httpx',
55 'websockets': 'websockets',
56 'jinja2': 'jinja2',
57 'flask': 'flask',
58 'django': 'django',
59 'boto3': 'boto3',
60 'openai': 'openai',
61 'anthropic': 'anthropic',
62 'langchain': 'langchain',
63 'transformers': 'transformers',
64 'torch': 'torch',
65 'tensorflow': 'tensorflow',
66 'sklearn': 'scikit-learn',
67 'cv2': 'opencv-python',
68 'PIL': 'Pillow',
69 'yaml': 'PyYAML',
70 'toml': 'toml',
71 'dotenv': 'python-dotenv',
72 'jwt': 'PyJWT',
73 'bcrypt': 'bcrypt',
74 'cryptography': 'cryptography',
75 }
77 def analyze_file(self, file_path: str) -> List[MissingDependency]:
78 """Analyze a Python file for missing dependencies."""
79 missing_deps = []
81 try:
82 with open(file_path, 'r', encoding='utf-8') as f:
83 content = f.read()
84 lines = content.splitlines()
86 # Parse the AST
87 tree = ast.parse(content)
89 # Find all import statements
90 for node in ast.walk(tree):
91 if isinstance(node, ast.Import):
92 for alias in node.names:
93 missing = self._check_module(alias.name, "import", node.lineno, lines)
94 if missing:
95 missing_deps.append(missing)
97 elif isinstance(node, ast.ImportFrom):
98 if node.module:
99 missing = self._check_module(node.module, "from_import", node.lineno, lines)
100 if missing:
101 missing_deps.append(missing)
103 except Exception as e:
104 logger.debug(f"Failed to analyze imports in {file_path}: {e}")
106 return missing_deps
108 def _check_module(self, module_name: str, import_type: str, line_num: int, lines: List[str]) -> Optional[MissingDependency]:
109 """Check if a module is available."""
110 # Get the top-level module name
111 top_module = module_name.split('.')[0]
113 # Skip standard library modules (basic check)
114 if self._is_stdlib_module(top_module):
115 return None
117 try:
118 importlib.import_module(top_module)
119 return None # Module is available
120 except ImportError:
121 source_line = lines[line_num - 1] if line_num <= len(lines) else None
123 # Check if this is an optional import (in try/except block)
124 # Convert AST line number (1-indexed) to list index (0-indexed)
125 is_optional = self._is_optional_import(lines, line_num - 1)
127 return MissingDependency(
128 module=top_module,
129 import_type="optional" if is_optional else import_type,
130 line_number=line_num,
131 source_line=source_line.strip() if source_line else None,
132 suggested_install=self.PIP_MAPPINGS.get(top_module, top_module)
133 )
135 def _is_stdlib_module(self, module_name: str) -> bool:
136 """Check if a module is part of the standard library."""
137 stdlib_modules = {
138 'os', 'sys', 'json', 'ast', 'inspect', 'importlib', 'pathlib',
139 'typing', 'logging', 'asyncio', 'traceback', 'dataclasses',
140 'functools', 'itertools', 'collections', 'datetime', 'time',
141 'math', 'random', 'string', 're', 'urllib', 'http', 'email',
142 'xml', 'html', 'sqlite3', 'pickle', 'csv', 'configparser',
143 'argparse', 'subprocess', 'threading', 'multiprocessing',
144 'queue', 'socket', 'ssl', 'hashlib', 'uuid', 'tempfile',
145 'shutil', 'glob', 'fnmatch', 'zipfile', 'tarfile', 'gzip',
146 'io', 'contextlib', 'warnings', 'unittest', 'doctest'
147 }
148 return module_name in stdlib_modules
150 def _is_optional_import(self, lines: List[str], line_index: int) -> bool:
151 """Check if an import is inside a try/except block."""
152 if line_index < 0 or line_index >= len(lines):
153 return False
155 # Get the indentation level of the import line
156 import_line = lines[line_index]
157 import_indent = len(import_line) - len(import_line.lstrip())
159 # Look backwards for 'try:' at the same or lesser indentation
160 try_found = False
161 for i in range(line_index - 1, max(-1, line_index - 20), -1):
162 line = lines[i]
163 if not line.strip(): # Skip empty lines
164 continue
166 line_indent = len(line) - len(line.lstrip())
168 # If we find a line at the same or lesser indentation that's not try:, stop
169 if line_indent <= import_indent:
170 if line.strip().startswith('try:'):
171 try_found = True
172 break
173 else:
174 # Found other code at same level, not in a try block
175 break
177 if not try_found:
178 return False
180 # Look forwards for 'except' block at the same indentation as try (less than import)
181 for i in range(line_index + 1, min(len(lines), line_index + 20)):
182 line = lines[i]
183 if not line.strip(): # Skip empty lines
184 continue
186 line_indent = len(line) - len(line.lstrip())
188 # Look for except at the same level as the try (less indented than the import)
189 if line_indent < import_indent:
190 if line.strip().startswith('except'):
191 return True
192 else:
193 # Found other code at try level, not an except block
194 break
195 # Continue looking through same-level code inside the try block
197 return False
200class MissingDependencyError(ImportError):
201 """Raised when required dependencies are missing."""
203 def __init__(self, message: str, missing_dependencies: List[MissingDependency]):
204 super().__init__(message)
205 self.missing_dependencies = missing_dependencies
207 def format_error_message(self) -> str:
208 """Format a detailed error message with installation suggestions."""
209 lines = [str(self)]
211 if not self.missing_dependencies:
212 return lines[0]
214 lines.append("\nMissing dependencies:")
216 # Group by required vs optional
217 required = [dep for dep in self.missing_dependencies if dep.import_type != "optional"]
218 optional = [dep for dep in self.missing_dependencies if dep.import_type == "optional"]
220 if required:
221 lines.append("\nRequired dependencies (will cause import errors):")
222 for dep in required:
223 lines.append(f" • {dep.module}")
224 if dep.source_line:
225 lines.append(f" Line {dep.line_number}: {dep.source_line}")
226 if dep.suggested_install and dep.suggested_install != dep.module:
227 lines.append(f" Install: pip install {dep.suggested_install}")
228 else:
229 lines.append(f" Install: pip install {dep.module}")
231 if optional:
232 lines.append("\nOptional dependencies (handled gracefully):")
233 for dep in optional:
234 lines.append(f" • {dep.module} (optional)")
235 if dep.suggested_install and dep.suggested_install != dep.module:
236 lines.append(f" Install: pip install {dep.suggested_install}")
237 else:
238 lines.append(f" Install: pip install {dep.module}")
240 # Generate installation command
241 required_installs = [dep.suggested_install or dep.module for dep in required]
242 if required_installs:
243 lines.append(f"\nQuick install: pip install {' '.join(required_installs)}")
245 return "\n".join(lines)
248class MCPFactory:
249 """Factory for creating MCP servers from Python code."""
251 def __init__(self, name: Optional[str] = None, version: str = "1.0.0", check_dependencies: bool = True):
252 """
253 Initialize the MCP factory.
255 Args:
256 name: Server name (defaults to module name)
257 version: Server version
258 check_dependencies: Whether to analyze and report missing dependencies
259 """
260 self.name = name
261 self.version = version
262 self.server: Optional[QuickMCPServer] = None
263 self.dependency_checking_enabled = check_dependencies
264 self.import_analyzer = ImportAnalyzer()
266 def from_module(self, module_path: str, include_private: bool = False) -> QuickMCPServer:
267 """
268 Create an MCP server from a Python module.
270 Args:
271 module_path: Path to Python file or module name
272 include_private: Include private functions (starting with _)
274 Returns:
275 QuickMCPServer with tools generated from module functions
277 Raises:
278 MissingDependencyError: If required dependencies are missing
279 """
280 # Analyze dependencies before attempting to load
281 missing_deps = []
282 if self.dependency_checking_enabled and Path(module_path).exists():
283 missing_deps = self.import_analyzer.analyze_file(module_path)
285 # Try to load the module
286 try:
287 module = self._load_module(module_path)
288 except ImportError as e:
289 # If we have dependency analysis, provide a detailed error
290 if missing_deps:
291 required_deps = [dep for dep in missing_deps if dep.import_type != "optional"]
292 if required_deps:
293 raise MissingDependencyError(
294 f"Failed to load module '{module_path}' due to missing dependencies",
295 missing_deps
296 ) from e
297 raise # Re-raise original error if no dependency analysis available
299 module_name = module.__name__ if hasattr(module, '__name__') else Path(module_path).stem
301 # Create server
302 server_name = self.name or f"{module_name}-mcp"
303 server_description = module.__doc__ or f"MCP server for {module_name}"
305 self.server = QuickMCPServer(
306 name=server_name,
307 version=self.version,
308 description=server_description.strip()
309 )
311 # Extract and register functions as tools
312 functions = self._extract_functions(module, include_private)
313 for func_name, func in functions.items():
314 self._register_function_as_tool(func_name, func)
316 logger.info(f"Created MCP server '{server_name}' with {len(functions)} tools")
318 # Report optional dependencies if any were found
319 if missing_deps:
320 optional_deps = [dep for dep in missing_deps if dep.import_type == "optional"]
321 if optional_deps:
322 logger.info(f"Note: {len(optional_deps)} optional dependencies not installed: " +
323 ", ".join([dep.module for dep in optional_deps]))
325 return self.server
327 def from_functions(self, functions: Dict[str, Callable], name: Optional[str] = None) -> QuickMCPServer:
328 """
329 Create an MCP server from a dictionary of functions.
331 Args:
332 functions: Dictionary mapping tool names to functions
333 name: Server name
335 Returns:
336 QuickMCPServer with specified functions as tools
337 """
338 server_name = name or self.name or "custom-mcp"
340 self.server = QuickMCPServer(
341 name=server_name,
342 version=self.version,
343 description=f"MCP server with {len(functions)} custom tools"
344 )
346 for func_name, func in functions.items():
347 self._register_function_as_tool(func_name, func)
349 return self.server
351 def from_class(self, cls: type, include_private: bool = False) -> QuickMCPServer:
352 """
353 Create an MCP server from a class.
355 Args:
356 cls: Class to extract methods from
357 include_private: Include private methods
359 Returns:
360 QuickMCPServer with class methods as tools
361 """
362 class_name = cls.__name__
363 server_name = self.name or f"{class_name.lower()}-mcp"
365 self.server = QuickMCPServer(
366 name=server_name,
367 version=self.version,
368 description=cls.__doc__ or f"MCP server for {class_name}"
369 )
371 # Create an instance if needed
372 instance = cls()
374 # Extract methods
375 for name in dir(instance):
376 if name.startswith('__'):
377 continue
378 if not include_private and name.startswith('_'):
379 continue
381 attr = getattr(instance, name)
382 if callable(attr) and not inspect.isclass(attr):
383 # Create a wrapper that captures the instance
384 def make_wrapper(method):
385 # Check if method is async
386 if inspect.iscoroutinefunction(method):
387 async def async_wrapper(**kwargs):
388 return await method(**kwargs)
389 async_wrapper.__name__ = method.__name__
390 async_wrapper.__doc__ = method.__doc__
391 async_wrapper.__annotations__ = getattr(method, '__annotations__', {})
392 return async_wrapper
393 else:
394 def wrapper(**kwargs):
395 return method(**kwargs)
396 wrapper.__name__ = method.__name__
397 wrapper.__doc__ = method.__doc__
398 wrapper.__annotations__ = getattr(method, '__annotations__', {})
399 return wrapper
401 self._register_function_as_tool(name, make_wrapper(attr))
403 return self.server
405 def from_file_with_decorators(self, file_path: str, decorator_name: str = "mcp_tool") -> QuickMCPServer:
406 """
407 Create an MCP server from functions decorated with a specific decorator.
409 Args:
410 file_path: Path to Python file
411 decorator_name: Name of decorator to look for
413 Returns:
414 QuickMCPServer with decorated functions as tools
415 """
416 # Parse the file to find decorated functions
417 with open(file_path, 'r') as f:
418 tree = ast.parse(f.read())
420 decorated_functions = []
421 for node in ast.walk(tree):
422 # Check both FunctionDef and AsyncFunctionDef
423 if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
424 for decorator in node.decorator_list:
425 if isinstance(decorator, ast.Name) and decorator.id == decorator_name:
426 decorated_functions.append(node.name)
427 elif isinstance(decorator, ast.Call):
428 if isinstance(decorator.func, ast.Name) and decorator.func.id == decorator_name:
429 decorated_functions.append(node.name)
431 # Load the module and extract decorated functions
432 module = self._load_module(file_path)
433 module_name = Path(file_path).stem
435 server_name = self.name or f"{module_name}-mcp"
436 self.server = QuickMCPServer(
437 name=server_name,
438 version=self.version,
439 description=f"MCP server for {module_name}"
440 )
442 for func_name in decorated_functions:
443 if hasattr(module, func_name):
444 func = getattr(module, func_name)
445 self._register_function_as_tool(func_name, func)
447 return self.server
449 def analyze_dependencies(self, module_path: str) -> List[MissingDependency]:
450 """
451 Analyze dependencies of a Python file without loading it.
453 Args:
454 module_path: Path to Python file
456 Returns:
457 List of missing dependencies
458 """
459 if not Path(module_path).exists():
460 return []
461 return self.import_analyzer.analyze_file(module_path)
463 def check_dependencies(self, module_path: str) -> Dict[str, Any]:
464 """
465 Check dependencies and return a summary report.
467 Args:
468 module_path: Path to Python file
470 Returns:
471 Dictionary with dependency analysis results
472 """
473 missing_deps = self.analyze_dependencies(module_path)
475 required = [dep for dep in missing_deps if dep.import_type != "optional"]
476 optional = [dep for dep in missing_deps if dep.import_type == "optional"]
478 return {
479 "file": module_path,
480 "total_missing": len(missing_deps),
481 "required_missing": len(required),
482 "optional_missing": len(optional),
483 "required_dependencies": required,
484 "optional_dependencies": optional,
485 "can_load": len(required) == 0,
486 "install_command": self._generate_install_command(required) if required else None
487 }
489 def _generate_install_command(self, dependencies: List[MissingDependency]) -> str:
490 """Generate a pip install command for missing dependencies."""
491 packages = []
492 for dep in dependencies:
493 packages.append(dep.suggested_install or dep.module)
494 return f"pip install {' '.join(set(packages))}" # Remove duplicates
496 def _load_module(self, module_path: str):
497 """Load a Python module from a file path or module name."""
498 path = Path(module_path)
500 if path.exists() and path.suffix == '.py':
501 # Load from file
502 spec = importlib.util.spec_from_file_location(path.stem, path)
503 if spec and spec.loader:
504 module = importlib.util.module_from_spec(spec)
505 sys.modules[path.stem] = module
506 spec.loader.exec_module(module)
507 return module
508 else:
509 # Try to import as module name
510 return importlib.import_module(module_path)
512 def _extract_functions(self, module, include_private: bool = False) -> Dict[str, Callable]:
513 """Extract functions from a module."""
514 functions = {}
516 for name in dir(module):
517 # Skip special attributes
518 if name.startswith('__'):
519 continue
521 # Skip private functions if requested
522 if not include_private and name.startswith('_'):
523 continue
525 attr = getattr(module, name)
527 # Only include functions (not classes, imports, etc.)
528 if callable(attr) and not inspect.isclass(attr):
529 # Check if it's defined in this module (not imported)
530 if hasattr(attr, '__module__'):
531 if attr.__module__ == module.__name__:
532 functions[name] = attr
533 else:
534 # Include if no module info (likely defined in this file)
535 functions[name] = attr
537 return functions
539 def _register_function_as_tool(self, name: str, func: Callable):
540 """Register a function as an MCP tool."""
541 if not self.server:
542 raise ValueError("Server not initialized")
544 # Get function signature and documentation
545 sig = inspect.signature(func)
546 doc = func.__doc__ or f"Execute {name}"
548 # Try to get type hints
549 try:
550 type_hints = get_type_hints(func)
551 except:
552 type_hints = {}
554 # Check if function is async
555 is_async = inspect.iscoroutinefunction(func)
557 # Create a wrapper that handles the conversion
558 if is_async:
559 async def tool_wrapper(**kwargs):
560 # Convert arguments to appropriate types if needed
561 converted_args = {}
562 for param_name, param in sig.parameters.items():
563 if param_name in kwargs:
564 value = kwargs[param_name]
565 # Try to convert based on type hints
566 if param_name in type_hints:
567 try:
568 param_type = type_hints[param_name]
569 if param_type != Any and not isinstance(value, param_type):
570 value = param_type(value)
571 except:
572 pass
573 converted_args[param_name] = value
574 elif param.kind == inspect.Parameter.VAR_KEYWORD:
575 # Handle **kwargs
576 for k, v in kwargs.items():
577 if k not in converted_args:
578 converted_args[k] = v
580 # Call the original function
581 result = await func(**converted_args)
583 # Convert result to JSON-serializable format
584 if result is None:
585 return {"success": True}
586 elif isinstance(result, (str, int, float, bool, list, dict)):
587 return result
588 else:
589 # Try to convert to dict or string
590 if hasattr(result, '__dict__'):
591 return result.__dict__
592 else:
593 return str(result)
594 else:
595 def tool_wrapper(**kwargs):
596 # Convert arguments to appropriate types if needed
597 converted_args = {}
598 for param_name, param in sig.parameters.items():
599 if param_name in kwargs:
600 value = kwargs[param_name]
601 # Try to convert based on type hints
602 if param_name in type_hints:
603 try:
604 param_type = type_hints[param_name]
605 if param_type != Any and not isinstance(value, param_type):
606 value = param_type(value)
607 except:
608 pass
609 converted_args[param_name] = value
610 elif param.kind == inspect.Parameter.VAR_KEYWORD:
611 # Handle **kwargs
612 for k, v in kwargs.items():
613 if k not in converted_args:
614 converted_args[k] = v
616 # Call the original function
617 result = func(**converted_args)
619 # Convert result to JSON-serializable format
620 if result is None:
621 return {"success": True}
622 elif isinstance(result, (str, int, float, bool, list, dict)):
623 return result
624 else:
625 # Try to convert to dict or string
626 if hasattr(result, '__dict__'):
627 return result.__dict__
628 else:
629 return str(result)
631 # Copy metadata
632 tool_wrapper.__name__ = name
633 tool_wrapper.__doc__ = doc
634 tool_wrapper.__annotations__ = func.__annotations__
636 # Register with the server
637 self.server.add_tool_from_function(tool_wrapper, name=name, description=doc.strip() if doc else f"Execute {name}")
640def create_mcp_from_module(
641 module_path: str,
642 server_name: Optional[str] = None,
643 include_private: bool = False,
644 auto_run: bool = True,
645 check_dependencies: bool = True
646) -> QuickMCPServer:
647 """
648 Convenience function to create and optionally run an MCP server from a module.
650 Args:
651 module_path: Path to Python file or module name
652 server_name: Optional server name
653 include_private: Include private functions
654 auto_run: Automatically run the server
655 check_dependencies: Whether to analyze and report missing dependencies
657 Returns:
658 QuickMCPServer instance
660 Raises:
661 MissingDependencyError: If required dependencies are missing
663 Example:
664 ```python
665 # Create MCP server from a Python file
666 server = create_mcp_from_module("my_utils.py")
668 # Or from an installed module
669 server = create_mcp_from_module("numpy", server_name="numpy-mcp")
671 # Check dependencies without loading
672 try:
673 server = create_mcp_from_module("my_utils.py")
674 except MissingDependencyError as e:
675 print(e.format_error_message())
676 ```
677 """
678 factory = MCPFactory(name=server_name, check_dependencies=check_dependencies)
679 server = factory.from_module(module_path, include_private=include_private)
681 if auto_run:
682 server.run()
684 return server
687def create_mcp_from_object(obj: Any, server_name: Optional[str] = None) -> QuickMCPServer:
688 """
689 Create an MCP server from any Python object.
691 Args:
692 obj: Python object (module, class, instance, or dict of functions)
693 server_name: Optional server name
695 Returns:
696 QuickMCPServer instance
697 """
698 factory = MCPFactory(name=server_name)
700 if isinstance(obj, dict):
701 # Dictionary of functions
702 return factory.from_functions(obj, name=server_name)
703 elif inspect.isclass(obj):
704 # Class
705 return factory.from_class(obj)
706 elif inspect.ismodule(obj):
707 # Module
708 return factory.from_module(obj.__file__ or obj.__name__)
709 elif hasattr(obj, '__class__') and obj.__class__.__module__ != 'builtins':
710 # Instance of a custom class (not built-in types)
711 return factory.from_class(obj.__class__)
712 else:
713 raise ValueError(f"Cannot create MCP server from {type(obj).__name__} object")
716# Decorator for marking functions to be exposed as MCP tools
717def mcp_tool(func: Optional[Callable] = None, *, name: Optional[str] = None, description: Optional[str] = None):
718 """
719 Decorator to mark a function for MCP exposure.
721 Args:
722 func: Function to decorate
723 name: Optional tool name (defaults to function name)
724 description: Optional description (defaults to docstring)
726 Example:
727 ```python
728 @mcp_tool
729 def calculate(x: int, y: int) -> int:
730 return x + y
732 @mcp_tool(name="custom_name", description="Custom description")
733 def my_function():
734 pass
735 ```
736 """
737 def decorator(f):
738 f._mcp_tool = True
739 f._mcp_name = name or f.__name__
740 f._mcp_description = description or f.__doc__
741 return f
743 if func is None:
744 return decorator
745 else:
746 return decorator(func)
749# Standalone utility functions for dependency analysis
750def analyze_dependencies(file_path: str) -> List[MissingDependency]:
751 """
752 Analyze dependencies of a Python file.
754 Args:
755 file_path: Path to Python file
757 Returns:
758 List of missing dependencies
759 """
760 analyzer = ImportAnalyzer()
761 return analyzer.analyze_file(file_path)
764def check_dependencies(file_path: str) -> Dict[str, Any]:
765 """
766 Check dependencies and return a detailed report.
768 Args:
769 file_path: Path to Python file
771 Returns:
772 Dictionary with dependency analysis results
773 """
774 factory = MCPFactory(check_dependencies=True)
775 return factory.check_dependencies(file_path)
778def print_dependency_report(file_path: str) -> None:
779 """
780 Print a formatted dependency report for a Python file.
782 Args:
783 file_path: Path to Python file
784 """
785 report = check_dependencies(file_path)
787 print(f"\nDependency Analysis for: {file_path}")
788 print("=" * 60)
790 if report["total_missing"] == 0:
791 print("✅ All dependencies are available!")
792 return
794 print(f"Total missing: {report['total_missing']}")
795 print(f"Required missing: {report['required_missing']}")
796 print(f"Optional missing: {report['optional_missing']}")
797 print(f"Can load module: {'✅ Yes' if report['can_load'] else '❌ No'}")
799 if report["required_dependencies"]:
800 print("\n❌ Required dependencies (will cause import errors):")
801 for dep in report["required_dependencies"]:
802 print(f" • {dep.module}")
803 if dep.source_line:
804 print(f" Line {dep.line_number}: {dep.source_line}")
805 install_pkg = dep.suggested_install or dep.module
806 if install_pkg != dep.module:
807 print(f" Install: pip install {install_pkg}")
808 else:
809 print(f" Install: pip install {dep.module}")
811 if report["optional_dependencies"]:
812 print("\n⚠️ Optional dependencies (handled gracefully):")
813 for dep in report["optional_dependencies"]:
814 print(f" • {dep.module}")
815 install_pkg = dep.suggested_install or dep.module
816 if install_pkg != dep.module:
817 print(f" Install: pip install {install_pkg}")
818 else:
819 print(f" Install: pip install {dep.module}")
821 if report["install_command"]:
822 print(f"\n💡 Quick install command:")
823 print(f" {report['install_command']}")
825 print()
828def get_install_command(file_path: str) -> Optional[str]:
829 """
830 Get pip install command for missing required dependencies.
832 Args:
833 file_path: Path to Python file
835 Returns:
836 Pip install command or None if no required dependencies are missing
837 """
838 report = check_dependencies(file_path)
839 return report.get("install_command")