Coverage for src/quickmcp/factory/core.py: 91%
204 statements
« prev ^ index » next coverage.py v7.10.4, created at 2025-08-20 20:56 +0200
1"""
2Core MCP Factory classes for generating 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
11import logging
13from ..server import QuickMCPServer
14from .config import FactoryConfig, DEFAULT_CONFIG
15from .import_analyzer import ImportAnalyzer, MissingDependency
16from .wrappers import ToolWrapperFactory, MethodToolWrapper
17from .errors import (
18 MissingDependencyError, ModuleLoadError, FunctionExtractionError,
19 CodeExecutionError, ToolRegistrationError, log_factory_error
20)
22logger = logging.getLogger(__name__)
25class ModuleLoader:
26 """Handles safe loading of Python modules."""
28 def __init__(self, config: Optional[FactoryConfig] = None):
29 self.config = config or DEFAULT_CONFIG
31 def load_module(self, module_path: str):
32 """Load a Python module from a file path or module name."""
33 if not self.config.allow_code_execution:
34 raise CodeExecutionError(
35 "Code execution is disabled. Enable 'allow_code_execution' in config to load modules.",
36 module_path
37 )
39 if self.config.warn_on_code_execution:
40 logger.warning(f"Loading module '{module_path}' will execute its code")
42 path = Path(module_path)
44 try:
45 if path.exists() and path.suffix == '.py':
46 # Load from file
47 spec = importlib.util.spec_from_file_location(path.stem, path)
48 if spec and spec.loader:
49 module = importlib.util.module_from_spec(spec)
50 sys.modules[path.stem] = module
51 spec.loader.exec_module(module)
52 return module
53 else:
54 raise ModuleLoadError(f"Could not create module spec for {path}", str(path))
55 else:
56 # Try to import as module name
57 return importlib.import_module(module_path)
59 except ImportError as e:
60 raise ModuleLoadError(f"Failed to import module: {e}", module_path, e)
61 except Exception as e:
62 raise ModuleLoadError(f"Error loading module: {e}", module_path, e)
65class FunctionExtractor:
66 """Extracts functions from modules and classes."""
68 def __init__(self, config: Optional[FactoryConfig] = None):
69 self.config = config or DEFAULT_CONFIG
71 def extract_from_module(self, module, include_private: bool = False) -> Dict[str, Callable]:
72 """Extract functions from a module."""
73 functions = {}
75 try:
76 for name in dir(module):
77 # Skip special attributes
78 if name.startswith('__'):
79 continue
81 # Skip private functions if requested
82 if not include_private and name.startswith('_'):
83 continue
85 attr = getattr(module, name)
87 # Only include functions (not classes, imports, etc.)
88 if callable(attr) and not inspect.isclass(attr):
89 # Check if it's defined in this module (not imported)
90 if hasattr(attr, '__module__'):
91 if attr.__module__ == module.__name__:
92 functions[name] = attr
93 else:
94 # Include if no module info (likely defined in this file)
95 functions[name] = attr
97 logger.info(f"Extracted {len(functions)} functions from module")
98 return functions
100 except Exception as e:
101 raise FunctionExtractionError(f"Failed to extract functions from module: {e}") from e
103 def extract_from_class(self, cls: type, include_private: bool = False) -> Dict[str, Callable]:
104 """Extract methods from a class."""
105 methods = {}
107 try:
108 # Create an instance if needed
109 instance = cls()
111 for name in dir(instance):
112 if name.startswith('__'):
113 continue
114 if not include_private and name.startswith('_'):
115 continue
117 attr = getattr(instance, name)
118 if callable(attr) and not inspect.isclass(attr):
119 methods[name] = attr
121 logger.info(f"Extracted {len(methods)} methods from class {cls.__name__}")
122 return methods, instance
124 except Exception as e:
125 raise FunctionExtractionError(f"Failed to extract methods from class {cls.__name__}: {e}") from e
127 def extract_decorated_functions(self, file_path: str, decorator_name: str = "mcp_tool") -> Dict[str, Callable]:
128 """Extract functions decorated with a specific decorator."""
129 try:
130 # Parse the file to find decorated functions
131 with open(file_path, 'r') as f:
132 tree = ast.parse(f.read())
134 decorated_functions = []
135 for node in ast.walk(tree):
136 if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
137 for decorator in node.decorator_list:
138 if isinstance(decorator, ast.Name) and decorator.id == decorator_name:
139 decorated_functions.append(node.name)
140 elif isinstance(decorator, ast.Call):
141 if isinstance(decorator.func, ast.Name) and decorator.func.id == decorator_name:
142 decorated_functions.append(node.name)
144 # Load the module and extract decorated functions
145 loader = ModuleLoader(self.config)
146 module = loader.load_module(file_path)
148 functions = {}
149 for func_name in decorated_functions:
150 if hasattr(module, func_name):
151 functions[func_name] = getattr(module, func_name)
153 logger.info(f"Extracted {len(functions)} decorated functions")
154 return functions
156 except Exception as e:
157 raise FunctionExtractionError(f"Failed to extract decorated functions: {e}") from e
160class MCPFactory:
161 """Factory for creating MCP servers from Python code."""
163 def __init__(self, name: Optional[str] = None, version: str = "1.0.0", config: Optional[FactoryConfig] = None):
164 """
165 Initialize the MCP factory.
167 Args:
168 name: Server name (defaults to module name)
169 version: Server version
170 config: Factory configuration
171 """
172 self.name = name
173 self.version = version
174 self.config = config or DEFAULT_CONFIG
176 # Initialize components
177 self.import_analyzer = ImportAnalyzer(self.config)
178 self.module_loader = ModuleLoader(self.config)
179 self.function_extractor = FunctionExtractor(self.config)
180 self.wrapper_factory = ToolWrapperFactory(self.config)
182 self.server: Optional[QuickMCPServer] = None
184 def from_module(self, module_path: str, include_private: bool = False) -> QuickMCPServer:
185 """
186 Create an MCP server from a Python module.
188 Args:
189 module_path: Path to Python file or module name
190 include_private: Include private functions (starting with _)
192 Returns:
193 QuickMCPServer with tools generated from module functions
195 Raises:
196 MissingDependencyError: If required dependencies are missing
197 """
198 # Analyze dependencies before attempting to load
199 missing_deps = []
200 if self.config.check_dependencies and Path(module_path).exists():
201 try:
202 missing_deps = self.import_analyzer.analyze_file(module_path)
203 except Exception as e:
204 logger.warning(f"Dependency analysis failed: {e}")
206 # Check for required missing dependencies
207 if missing_deps:
208 required_deps = [dep for dep in missing_deps if dep.import_type != "optional"]
209 if required_deps:
210 raise MissingDependencyError(
211 f"Cannot load module '{module_path}' due to missing dependencies",
212 missing_deps, module_path
213 )
215 # Load the module
216 try:
217 module = self.module_loader.load_module(module_path)
218 except ModuleLoadError as e:
219 # If we have dependency analysis, enhance the error
220 if missing_deps:
221 raise MissingDependencyError(
222 f"Failed to load module '{module_path}' due to missing dependencies",
223 missing_deps, module_path
224 ) from e
225 raise
227 # Extract functions
228 functions = self.function_extractor.extract_from_module(module, include_private)
230 # Create server
231 module_name = module.__name__ if hasattr(module, '__name__') else Path(module_path).stem
232 server_name = self.name or f"{module_name}-mcp"
233 server_description = module.__doc__ or f"MCP server for {module_name}"
235 self.server = QuickMCPServer(
236 name=server_name,
237 version=self.version,
238 description=server_description.strip()
239 )
241 # Register functions as tools
242 for func_name, func in functions.items():
243 self._register_tool(func_name, func)
245 # Report optional dependencies
246 if missing_deps and self.config.warn_on_optional_missing:
247 optional_deps = [dep for dep in missing_deps if dep.import_type == "optional"]
248 if optional_deps:
249 logger.info(f"Note: {len(optional_deps)} optional dependencies not installed: " +
250 ", ".join([dep.module for dep in optional_deps]))
252 logger.info(f"Created MCP server '{server_name}' with {len(functions)} tools")
253 return self.server
255 def from_module_object(self, module, include_private: bool = False) -> QuickMCPServer:
256 """
257 Create an MCP server from an already-loaded module object.
259 Args:
260 module: Already-loaded module object
261 include_private: Include private functions
263 Returns:
264 QuickMCPServer with module functions as tools
265 """
266 # Extract functions from the module
267 functions = self.function_extractor.extract_from_module(module, include_private)
269 # Create server
270 module_name = module.__name__ if hasattr(module, '__name__') else "module"
271 server_name = self.name or f"{module_name.replace('.', '-')}-mcp"
272 server_description = module.__doc__ or f"MCP server for {module_name}"
274 self.server = QuickMCPServer(
275 name=server_name,
276 version=self.version,
277 description=server_description.strip() if server_description else ""
278 )
280 # Register functions as tools
281 for func_name, func in functions.items():
282 self._register_tool(func_name, func)
284 logger.info(f"Created MCP server '{server_name}' with {len(functions)} tools")
285 return self.server
287 def from_functions(self, functions: Dict[str, Callable], name: Optional[str] = None) -> QuickMCPServer:
288 """
289 Create an MCP server from a dictionary of functions.
291 Args:
292 functions: Dictionary mapping tool names to functions
293 name: Server name
295 Returns:
296 QuickMCPServer with specified functions as tools
297 """
298 server_name = name or self.name or "custom-mcp"
300 self.server = QuickMCPServer(
301 name=server_name,
302 version=self.version,
303 description=f"MCP server with {len(functions)} custom tools"
304 )
306 for func_name, func in functions.items():
307 self._register_tool(func_name, func)
309 logger.info(f"Created MCP server '{server_name}' with {len(functions)} tools")
310 return self.server
312 def from_class(self, cls: type, include_private: bool = False) -> QuickMCPServer:
313 """
314 Create an MCP server from a class.
316 Args:
317 cls: Class to extract methods from
318 include_private: Include private methods
320 Returns:
321 QuickMCPServer with class methods as tools
322 """
323 class_name = cls.__name__
324 server_name = self.name or f"{class_name.lower()}-mcp"
326 self.server = QuickMCPServer(
327 name=server_name,
328 version=self.version,
329 description=cls.__doc__ or f"MCP server for {class_name}"
330 )
332 # Extract methods and instance
333 methods, instance = self.function_extractor.extract_from_class(cls, include_private)
335 # Register methods as tools
336 for method_name, method in methods.items():
337 try:
338 wrapper = self.wrapper_factory.create_method_wrapper(method, instance, method_name)
339 self.server.add_tool_from_function(
340 wrapper.wrapper,
341 name=method_name,
342 description=wrapper.doc.strip() if wrapper.doc else f"Execute {method_name}"
343 )
344 except Exception as e:
345 error = ToolRegistrationError(f"Failed to register method '{method_name}': {e}", method_name, e)
346 log_factory_error(error, f"Class {class_name}")
347 if self.config.strict_type_conversion:
348 raise error
350 logger.info(f"Created MCP server '{server_name}' with {len(methods)} method tools")
351 return self.server
353 def from_file_with_decorators(self, file_path: str, decorator_name: str = "mcp_tool") -> QuickMCPServer:
354 """
355 Create an MCP server from functions decorated with a specific decorator.
357 Args:
358 file_path: Path to Python file
359 decorator_name: Name of decorator to look for
361 Returns:
362 QuickMCPServer with decorated functions as tools
363 """
364 # Extract decorated functions
365 functions = self.function_extractor.extract_decorated_functions(file_path, decorator_name)
367 # Create server
368 module_name = Path(file_path).stem
369 server_name = self.name or f"{module_name}-mcp"
371 self.server = QuickMCPServer(
372 name=server_name,
373 version=self.version,
374 description=f"MCP server for {module_name}"
375 )
377 # Register functions as tools
378 for func_name, func in functions.items():
379 self._register_tool(func_name, func)
381 logger.info(f"Created MCP server '{server_name}' with {len(functions)} decorated tools")
382 return self.server
384 def _register_tool(self, name: str, func: Callable):
385 """Register a function as an MCP tool."""
386 if not self.server:
387 raise ToolRegistrationError("Server not initialized", name)
389 try:
390 wrapper = self.wrapper_factory.create_wrapper(func, name)
391 self.server.add_tool_from_function(
392 wrapper.wrapper,
393 name=name,
394 description=wrapper.doc.strip() if wrapper.doc else f"Execute {name}"
395 )
396 except Exception as e:
397 error = ToolRegistrationError(f"Failed to register tool '{name}': {e}", name, e)
398 log_factory_error(error, f"Tool registration")
399 if self.config.strict_type_conversion:
400 raise error
402 def analyze_dependencies(self, module_path: str) -> List[MissingDependency]:
403 """
404 Analyze dependencies of a Python file without loading it.
406 Args:
407 module_path: Path to Python file
409 Returns:
410 List of missing dependencies
411 """
412 if not Path(module_path).exists():
413 return []
414 return self.import_analyzer.analyze_file(module_path)
416 def check_dependencies(self, module_path: str) -> Dict[str, Any]:
417 """
418 Check dependencies and return a summary report.
420 Args:
421 module_path: Path to Python file
423 Returns:
424 Dictionary with dependency analysis results
425 """
426 missing_deps = self.analyze_dependencies(module_path)
428 required = [dep for dep in missing_deps if dep.import_type != "optional"]
429 optional = [dep for dep in missing_deps if dep.import_type == "optional"]
430 dev_deps = [dep for dep in missing_deps if dep.is_dev_dependency]
432 return {
433 "file": module_path,
434 "total_missing": len(missing_deps),
435 "required_missing": len(required),
436 "optional_missing": len(optional),
437 "dev_missing": len(dev_deps),
438 "required_dependencies": required,
439 "optional_dependencies": optional,
440 "dev_dependencies": dev_deps,
441 "can_load": len(required) == 0,
442 "install_command": self._generate_install_command(required) if required else None
443 }
445 def _generate_install_command(self, dependencies: List[MissingDependency]) -> str:
446 """Generate a pip install command for missing dependencies."""
447 packages = []
448 for dep in dependencies:
449 packages.append(dep.suggested_install or dep.module)
450 return f"pip install {' '.join(set(packages))}" # Remove duplicates