"""Shared preflight analysis engine. Produces a PreflightReport used by the GUI, CLI, and registry workflow. All checks are deterministic and evidence-based. """ from dataclasses import dataclass, field from typing import Any, Dict, List, Optional from .tool import Tool, validate_json_schema @dataclass class PreflightReport: """Structured analysis of a tool before creation, save, or publish.""" errors: List[str] = field(default_factory=list) warnings: List[str] = field(default_factory=list) suggestions: List[str] = field(default_factory=list) similar_tools: List[Dict[str, Any]] = field(default_factory=list) generated_tests: List[Dict[str, Any]] = field(default_factory=list) compatibility: List[Dict[str, Any]] = field(default_factory=list) @property def ok(self) -> bool: return not self.errors def merge(self, other: "PreflightReport") -> "PreflightReport": return PreflightReport( errors=self.errors + other.errors, warnings=self.warnings + other.warnings, suggestions=self.suggestions + other.suggestions, similar_tools=self.similar_tools + other.similar_tools, generated_tests=self.generated_tests + other.generated_tests, compatibility=self.compatibility + other.compatibility, ) def to_dict(self) -> Dict[str, Any]: return { "errors": list(self.errors), "warnings": list(self.warnings), "suggestions": list(self.suggestions), "similar_tools": list(self.similar_tools), "generated_tests": list(self.generated_tests), "compatibility": list(self.compatibility), } def analyze_tool( tool: Tool, registry_client=None, *, check_local_dependencies: bool = True, ) -> PreflightReport: """Run all preflight checks on a tool. Args: tool: The Tool to analyze. registry_client: Optional registry client for registry-side checks. Returns: A PreflightReport with structured findings. """ report = PreflightReport() _check_config_integrity(tool, report) _check_contracts(tool, report) _check_secrets(tool, report) if check_local_dependencies: _check_dependencies(tool, report) if registry_client: _check_similar_tools(tool, report, registry_client) return report def _check_config_integrity(tool: Tool, report: PreflightReport): if not tool.name: report.errors.append("Tool name is required") if tool.version and not _is_semver(tool.version): report.errors.append(f"Version '{tool.version}' is not valid semver") if not tool.steps and not tool.arguments and tool.output == "{input}": report.warnings.append( "Tool has no steps and no arguments — output template is passthrough" ) def _check_contracts(tool: Tool, report: PreflightReport): if tool.output_schema is not None: _validate_schema(tool.output_schema, "output_schema", report) if tool.input_schema is not None: _validate_schema(tool.input_schema, "input_schema", report) if tool.output_schema is None and tool.input_schema is None: report.suggestions.append( "Add input_schema and output_schema to enable automated verification" ) def _validate_schema(schema: Any, label: str, report: PreflightReport): try: validate_json_schema(schema, label) except ValueError as exc: report.errors.append(str(exc)) def _check_secrets(tool: Tool, report: PreflightReport): secret_patterns = ["api_key", "api_secret", "password", "token", "secret"] for step in tool.steps: if hasattr(step, "prompt"): prompt_lower = step.prompt.lower() for pat in secret_patterns: if pat in prompt_lower: report.warnings.append( f"Prompt step contains potential secret pattern '{pat}'" ) break def _check_dependencies(tool: Tool, report: PreflightReport): from .tool import tool_exists for dep in tool.dependencies: ref = dep.get("name", dep) if isinstance(dep, dict) else dep if not tool_exists(ref): report.warnings.append(f"Dependency '{ref}' is not installed locally") def _check_similar_tools(tool: Tool, report: PreflightReport, client): from .registry_client import RegistryError try: results = client.search_tools(tool.name, per_page=5) items = results.data if hasattr(results, "data") else results for item in items: report.similar_tools.append({ "name": (item.get("owner") or "") + "/" + item.get("name", ""), "description": item.get("description", ""), "downloads": item.get("downloads", 0), }) if not items: report.suggestions.append("No similar tools found on the registry") except RegistryError: report.warnings.append("Could not search registry for similar tools") def _is_semver(version: str) -> bool: from .semver import Version return Version.parse(version) is not None