snyk_scan_library
Scan libraries for security vulnerabilities using Snyk analysis. Identify risks, check licenses, and get remediation guidance for packages in ecosystems like npm, PyPI, or Maven.
Instructions
Scan a library using Snyk for comprehensive security analysis.
Args:
library_name: Name of the library to scan
version: Version of the library (default: "latest")
ecosystem: Package ecosystem ("pypi", "npm", "maven", etc.)
Returns:
Detailed security report from Snyk including vulnerabilities, licenses, and remediation adviceInput Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| library_name | Yes | ||
| version | No | latest | |
| ecosystem | No | pypi |
Implementation Reference
- The primary handler function for the 'snyk_scan_library' MCP tool. Registers the tool via @mcp.tool(), handles input parameters, performs Snyk connection test and package scan, formats the security report with vulnerabilities, licenses, and recommendations.
@mcp.tool() async def snyk_scan_library( library_name: str, version: str = "latest", ecosystem: str = "pypi" ): """ Scan a library using Snyk for comprehensive security analysis. Args: library_name: Name of the library to scan version: Version of the library (default: "latest") ecosystem: Package ecosystem ("pypi", "npm", "maven", etc.) Returns: Detailed security report from Snyk including vulnerabilities, licenses, and remediation advice """ from .snyk_integration import snyk_integration try: # Test connection first connection_test = await snyk_integration.test_connection() if connection_test["status"] != "connected": return { "error": "Snyk integration not configured", "details": connection_test.get("error", "Unknown error"), "setup_instructions": [ "1. Sign up for Snyk account at https://snyk.io", "2. Get API token from your Snyk account settings", "3. Set SNYK_API_KEY environment variable", "4. Optionally set SNYK_ORG_ID for organization-specific scans", ], } # Perform the scan package_info = await snyk_integration.scan_package( library_name, version, ecosystem ) return { "library": library_name, "version": version, "ecosystem": ecosystem, "scan_timestamp": datetime.now().isoformat(), "vulnerability_summary": { "total": len(package_info.vulnerabilities), "critical": package_info.severity_counts.get("critical", 0), "high": package_info.severity_counts.get("high", 0), "medium": package_info.severity_counts.get("medium", 0), "low": package_info.severity_counts.get("low", 0), }, "vulnerabilities": [ { "id": vuln.id, "title": vuln.title, "severity": vuln.severity.value, "cvss_score": vuln.cvss_score, "cve": vuln.cve, "is_patchable": vuln.is_patchable, "upgrade_path": vuln.upgrade_path[:3] if vuln.upgrade_path else [], "snyk_url": f"https://snyk.io/vuln/{vuln.id}", } for vuln in package_info.vulnerabilities[:10] # Limit to top 10 ], "license_info": [ { "name": license.name, "type": license.type, "spdx_id": license.spdx_id, "is_deprecated": license.is_deprecated, } for license in package_info.licenses ], "recommendations": [ "🔍 Review all critical and high severity vulnerabilities", "📦 Update to latest secure version if available", "⚖️ Ensure license compliance with your organization's policies", "🔄 Set up continuous monitoring for this package", ], } except Exception as e: return { "error": f"Snyk scan failed: {str(e)}", "library": library_name, "version": version, } - Core helper method that executes the Snyk API scan for a specific package/version/ecosystem, handles caching, creates mock manifest payloads, parses responses, and returns structured SnykPackageInfo.
async def scan_package( self, package_name: str, version: str, ecosystem: str = "pypi" ) -> SnykPackageInfo: """Scan a single package for vulnerabilities""" cache_key = f"package_{ecosystem}_{package_name}_{version}" if self._is_cached(cache_key): return self.cache[cache_key]["data"] ecosystem_map = { "pypi": "pip", "npm": "npm", "maven": "maven", "gradle": "gradle", "nuget": "nuget", } snyk_ecosystem = ecosystem_map.get(ecosystem.lower(), "pip") try: async with httpx.AsyncClient(timeout=self.timeout) as client: # Test endpoint for vulnerabilities test_payload = { "encoding": "plain", "files": { ( "requirements.txt" if snyk_ecosystem == "pip" else "package.json" ): { "contents": ( f"{package_name}=={version}" if snyk_ecosystem == "pip" else json.dumps( {"dependencies": {package_name: version}} ) ) } }, } response = await client.post( f"{self.base_url}/v1/test/{snyk_ecosystem}", headers=self._get_headers(), json=test_payload, ) if response.status_code == 200: data = response.json() package_info = self._parse_package_scan_result( data, package_name, version, ecosystem ) # Cache the result self._cache_result(cache_key, package_info) return package_info else: # Return empty result for failed scans return SnykPackageInfo( name=package_name, version=version, ecosystem=ecosystem, vulnerabilities=[], licenses=[], severity_counts={ "critical": 0, "high": 0, "medium": 0, "low": 0, }, dependency_paths=[], is_direct_dependency=True, ) except Exception as e: print(f"Snyk scan error for {package_name}: {e}", file=sys.stderr) return SnykPackageInfo( name=package_name, version=version, ecosystem=ecosystem, vulnerabilities=[], licenses=[], severity_counts={"critical": 0, "high": 0, "medium": 0, "low": 0}, dependency_paths=[], is_direct_dependency=True, ) - Global singleton instance of SnykIntegration used by the tool handler.
snyk_integration = SnykIntegration() - src/documentation_search_enhanced/main.py:1779-1779 (registration)MCP tool registration decorator for snyk_scan_library.
@mcp.tool() - Tool schema defined by function signature and docstring describing inputs (library_name, version, ecosystem) and output format.
library_name: str, version: str = "latest", ecosystem: str = "pypi" ): """ Scan a library using Snyk for comprehensive security analysis. Args: library_name: Name of the library to scan version: Version of the library (default: "latest") ecosystem: Package ecosystem ("pypi", "npm", "maven", etc.) Returns: Detailed security report from Snyk including vulnerabilities, licenses, and remediation advice """