MCP/pypi-query-mcp
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

feat: Complete PyPI Query MCP Server Implementation (#3)

Browse source 
Merge pull request implementing complete PyPI query MCP server with comprehensive features and CI/CD pipeline.
This commit is contained in:
Hal 2025-05-27 11:14:49 +08:00 • committed by GitHub
parent b1a1a6866d
commit 030b3a2607
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
33 changed files with 3238 additions and 246 deletions
Download patch file Download diff file Expand all files Collapse all files

25
pypi_query_mcp/tools/__init__.py Normal file
View file

@ -0,0 +1,25 @@
"""MCP tools for PyPI package queries.
This package contains the FastMCP tool implementations that provide
the user-facing interface for PyPI package operations.
"""
from .compatibility_check import (
check_python_compatibility,
get_compatible_python_versions,
suggest_python_version_for_packages,
)
from .package_query import (
query_package_dependencies,
query_package_info,
query_package_versions,
)
__all__ = [
"query_package_info",
"query_package_versions",
"query_package_dependencies",
"check_python_compatibility",
"get_compatible_python_versions",
"suggest_python_version_for_packages",
]

260
pypi_query_mcp/tools/compatibility_check.py Normal file
View file

@ -0,0 +1,260 @@
"""Python version compatibility checking tools for PyPI MCP server."""
import logging
from typing import Any
from ..core import (
InvalidPackageNameError,
NetworkError,
PyPIClient,
PyPIError,
)
from ..core.version_utils import VersionCompatibility
logger = logging.getLogger(__name__)
async def check_python_compatibility(
package_name: str,
target_python_version: str,
use_cache: bool = True
) -> dict[str, Any]:
"""Check if a package is compatible with a specific Python version.
Args:
package_name: Name of the package to check
target_python_version: Target Python version (e.g., "3.9", "3.10.5")
use_cache: Whether to use cached package data
Returns:
Dictionary containing compatibility information
Raises:
InvalidPackageNameError: If package name is invalid
PackageNotFoundError: If package is not found
NetworkError: For network-related errors
"""
if not package_name or not package_name.strip():
raise InvalidPackageNameError(package_name)
if not target_python_version or not target_python_version.strip():
raise ValueError("Target Python version cannot be empty")
logger.info(f"Checking Python {target_python_version} compatibility for package: {package_name}")
try:
async with PyPIClient() as client:
package_data = await client.get_package_info(package_name, use_cache)
info = package_data.get("info", {})
requires_python = info.get("requires_python")
classifiers = info.get("classifiers", [])
# Perform compatibility check
compat_checker = VersionCompatibility()
result = compat_checker.check_version_compatibility(
target_python_version,
requires_python,
classifiers
)
# Add package information to result
result.update({
"package_name": info.get("name", package_name),
"package_version": info.get("version", ""),
"requires_python": requires_python,
"supported_implementations": compat_checker.extract_python_implementations(classifiers),
"classifier_versions": sorted(compat_checker.extract_python_versions_from_classifiers(classifiers))
})
return result
except PyPIError:
# Re-raise PyPI-specific errors
raise
except Exception as e:
logger.error(f"Unexpected error checking compatibility for {package_name}: {e}")
raise NetworkError(f"Failed to check Python compatibility: {e}", e) from e
async def get_compatible_python_versions(
package_name: str,
python_versions: list[str] | None = None,
use_cache: bool = True
) -> dict[str, Any]:
"""Get list of Python versions compatible with a package.
Args:
package_name: Name of the package to check
python_versions: List of Python versions to check (optional)
use_cache: Whether to use cached package data
Returns:
Dictionary containing compatibility information for multiple versions
Raises:
InvalidPackageNameError: If package name is invalid
PackageNotFoundError: If package is not found
NetworkError: For network-related errors
"""
if not package_name or not package_name.strip():
raise InvalidPackageNameError(package_name)
logger.info(f"Getting compatible Python versions for package: {package_name}")
try:
async with PyPIClient() as client:
package_data = await client.get_package_info(package_name, use_cache)
info = package_data.get("info", {})
requires_python = info.get("requires_python")
classifiers = info.get("classifiers", [])
# Get compatibility information
compat_checker = VersionCompatibility()
result = compat_checker.get_compatible_versions(
requires_python,
classifiers,
python_versions
)
# Add package information to result
result.update({
"package_name": info.get("name", package_name),
"package_version": info.get("version", ""),
"requires_python": requires_python,
"supported_implementations": sorted(compat_checker.extract_python_implementations(classifiers)),
"classifier_versions": sorted(compat_checker.extract_python_versions_from_classifiers(classifiers))
})
return result
except PyPIError:
# Re-raise PyPI-specific errors
raise
except Exception as e:
logger.error(f"Unexpected error getting compatible versions for {package_name}: {e}")
raise NetworkError(f"Failed to get compatible Python versions: {e}", e) from e
async def suggest_python_version_for_packages(
package_names: list[str],
use_cache: bool = True
) -> dict[str, Any]:
"""Suggest optimal Python version for a list of packages.
Args:
package_names: List of package names to analyze
use_cache: Whether to use cached package data
Returns:
Dictionary containing version suggestions and compatibility matrix
Raises:
ValueError: If package_names is empty
NetworkError: For network-related errors
"""
if not package_names:
raise ValueError("Package names list cannot be empty")
logger.info(f"Analyzing Python version compatibility for {len(package_names)} packages")
# Default Python versions to analyze
python_versions = ["3.7", "3.8", "3.9", "3.10", "3.11", "3.12"]
compatibility_matrix = {}
package_details = {}
errors = {}
async with PyPIClient() as client:
for package_name in package_names:
try:
package_data = await client.get_package_info(package_name, use_cache)
info = package_data.get("info", {})
requires_python = info.get("requires_python")
classifiers = info.get("classifiers", [])
compat_checker = VersionCompatibility()
compat_result = compat_checker.get_compatible_versions(
requires_python,
classifiers,
python_versions
)
# Store compatibility for this package
compatible_versions = [v["version"] for v in compat_result["compatible_versions"]]
compatibility_matrix[package_name] = compatible_versions
package_details[package_name] = {
"version": info.get("version", ""),
"requires_python": requires_python,
"compatible_versions": compatible_versions,
"compatibility_rate": compat_result["compatibility_rate"]
}
except Exception as e:
logger.warning(f"Failed to analyze package {package_name}: {e}")
errors[package_name] = str(e)
compatibility_matrix[package_name] = []
# Find common compatible versions
if compatibility_matrix:
all_versions = set(python_versions)
common_versions = all_versions.copy()
for _package_name, compatible in compatibility_matrix.items():
if compatible: # Only consider packages with known compatibility
common_versions &= set(compatible)
else:
common_versions = set()
# Generate recommendations
recommendations = []
if common_versions:
latest_common = max(common_versions, key=lambda x: tuple(map(int, x.split("."))))
recommendations.append(
f"✅ Recommended Python version: {latest_common} "
f"(compatible with all {len([p for p in compatibility_matrix if compatibility_matrix[p]])} packages)"
)
if len(common_versions) > 1:
all_common = sorted(common_versions, key=lambda x: tuple(map(int, x.split("."))))
recommendations.append(
f"📋 All compatible versions: {', '.join(all_common)}"
)
else:
recommendations.append(
"⚠️ No Python version is compatible with all packages. "
"Consider updating packages or using different versions."
)
# Find the version compatible with most packages
version_scores = {}
for version in python_versions:
score = sum(1 for compatible in compatibility_matrix.values() if version in compatible)
version_scores[version] = score
if version_scores:
best_version = max(version_scores, key=version_scores.get)
best_score = version_scores[best_version]
total_packages = len([p for p in compatibility_matrix if compatibility_matrix[p]])
if best_score > 0:
recommendations.append(
f"📊 Best compromise: Python {best_version} "
f"(compatible with {best_score}/{total_packages} packages)"
)
return {
"analyzed_packages": len(package_names),
"successful_analyses": len(package_details),
"failed_analyses": len(errors),
"common_compatible_versions": sorted(common_versions),
"recommended_version": max(common_versions, key=lambda x: tuple(map(int, x.split(".")))) if common_versions else None,
"compatibility_matrix": compatibility_matrix,
"package_details": package_details,
"errors": errors,
"recommendations": recommendations,
"python_versions_analyzed": python_versions
}

253
pypi_query_mcp/tools/package_query.py Normal file
View file

@ -0,0 +1,253 @@
"""Package query tools for PyPI MCP server."""
import logging
from typing import Any
from ..core import (
InvalidPackageNameError,
NetworkError,
PyPIClient,
PyPIError,
)
logger = logging.getLogger(__name__)
def format_package_info(package_data: dict[str, Any]) -> dict[str, Any]:
"""Format package information for MCP response.
Args:
package_data: Raw package data from PyPI API
Returns:
Formatted package information
"""
info = package_data.get("info", {})
# Extract basic information
formatted = {
"name": info.get("name", ""),
"version": info.get("version", ""),
"summary": info.get("summary", ""),
"description": info.get("description", "")[:500] + "..." if len(info.get("description", "")) > 500 else info.get("description", ""),
"author": info.get("author", ""),
"author_email": info.get("author_email", ""),
"maintainer": info.get("maintainer", ""),
"maintainer_email": info.get("maintainer_email", ""),
"license": info.get("license", ""),
"home_page": info.get("home_page", ""),
"project_url": info.get("project_url", ""),
"download_url": info.get("download_url", ""),
"requires_python": info.get("requires_python", ""),
"platform": info.get("platform", ""),
"keywords": info.get("keywords", ""),
"classifiers": info.get("classifiers", []),
"requires_dist": info.get("requires_dist", []),
"project_urls": info.get("project_urls", {}),
}
# Add release information
releases = package_data.get("releases", {})
formatted["total_versions"] = len(releases)
formatted["available_versions"] = list(releases.keys())[-10:] # Last 10 versions
# Add download statistics if available
if "urls" in package_data:
urls = package_data["urls"]
if urls:
formatted["download_info"] = {
"files_count": len(urls),
"file_types": list({url.get("packagetype", "") for url in urls}),
"python_versions": list({url.get("python_version", "") for url in urls if url.get("python_version")}),
}
return formatted
def format_version_info(package_data: dict[str, Any]) -> dict[str, Any]:
"""Format version information for MCP response.
Args:
package_data: Raw package data from PyPI API
Returns:
Formatted version information
"""
info = package_data.get("info", {})
releases = package_data.get("releases", {})
# Sort versions (basic sorting, could be improved with proper version parsing)
sorted_versions = sorted(releases.keys(), reverse=True)
return {
"package_name": info.get("name", ""),
"latest_version": info.get("version", ""),
"total_versions": len(releases),
"versions": sorted_versions,
"recent_versions": sorted_versions[:20], # Last 20 versions
"version_details": {
version: {
"release_count": len(releases[version]),
"has_wheel": any(file.get("packagetype") == "bdist_wheel" for file in releases[version]),
"has_source": any(file.get("packagetype") == "sdist" for file in releases[version]),
}
for version in sorted_versions[:10] # Details for last 10 versions
}
}
def format_dependency_info(package_data: dict[str, Any]) -> dict[str, Any]:
"""Format dependency information for MCP response.
Args:
package_data: Raw package data from PyPI API
Returns:
Formatted dependency information
"""
info = package_data.get("info", {})
requires_dist = info.get("requires_dist", []) or []
# Parse dependencies
runtime_deps = []
dev_deps = []
optional_deps = {}
for dep in requires_dist:
if not dep:
continue
# Basic parsing - could be improved with proper dependency parsing
if "extra ==" in dep:
# Optional dependency
parts = dep.split(";")
dep_name = parts[0].strip()
extra_part = parts[1] if len(parts) > 1 else ""
if "extra ==" in extra_part:
extra_name = extra_part.split("extra ==")[1].strip().strip('"\'')
if extra_name not in optional_deps:
optional_deps[extra_name] = []
optional_deps[extra_name].append(dep_name)
elif "dev" in dep.lower() or "test" in dep.lower():
dev_deps.append(dep)
else:
runtime_deps.append(dep)
return {
"package_name": info.get("name", ""),
"version": info.get("version", ""),
"requires_python": info.get("requires_python", ""),
"runtime_dependencies": runtime_deps,
"development_dependencies": dev_deps,
"optional_dependencies": optional_deps,
"total_dependencies": len(requires_dist),
"dependency_summary": {
"runtime_count": len(runtime_deps),
"dev_count": len(dev_deps),
"optional_groups": len(optional_deps),
"total_optional": sum(len(deps) for deps in optional_deps.values()),
}
}
async def query_package_info(package_name: str) -> dict[str, Any]:
"""Query comprehensive package information from PyPI.
Args:
package_name: Name of the package to query
Returns:
Formatted package information
Raises:
InvalidPackageNameError: If package name is invalid
PackageNotFoundError: If package is not found
NetworkError: For network-related errors
"""
if not package_name or not package_name.strip():
raise InvalidPackageNameError(package_name)
logger.info(f"Querying package info for: {package_name}")
try:
async with PyPIClient() as client:
package_data = await client.get_package_info(package_name)
return format_package_info(package_data)
except PyPIError:
# Re-raise PyPI-specific errors
raise
except Exception as e:
logger.error(f"Unexpected error querying package {package_name}: {e}")
raise NetworkError(f"Failed to query package information: {e}", e) from e
async def query_package_versions(package_name: str) -> dict[str, Any]:
"""Query package version information from PyPI.
Args:
package_name: Name of the package to query
Returns:
Formatted version information
Raises:
InvalidPackageNameError: If package name is invalid
PackageNotFoundError: If package is not found
NetworkError: For network-related errors
"""
if not package_name or not package_name.strip():
raise InvalidPackageNameError(package_name)
logger.info(f"Querying versions for package: {package_name}")
try:
async with PyPIClient() as client:
package_data = await client.get_package_info(package_name)
return format_version_info(package_data)
except PyPIError:
# Re-raise PyPI-specific errors
raise
except Exception as e:
logger.error(f"Unexpected error querying versions for {package_name}: {e}")
raise NetworkError(f"Failed to query package versions: {e}", e) from e
async def query_package_dependencies(package_name: str, version: str | None = None) -> dict[str, Any]:
"""Query package dependency information from PyPI.
Args:
package_name: Name of the package to query
version: Specific version to query (optional, defaults to latest)
Returns:
Formatted dependency information
Raises:
InvalidPackageNameError: If package name is invalid
PackageNotFoundError: If package is not found
NetworkError: For network-related errors
"""
if not package_name or not package_name.strip():
raise InvalidPackageNameError(package_name)
logger.info(f"Querying dependencies for package: {package_name}" +
(f" version {version}" if version else " (latest)"))
try:
async with PyPIClient() as client:
package_data = await client.get_package_info(package_name)
# TODO: In future, support querying specific version dependencies
# For now, we return dependencies for the latest version
if version and version != package_data.get("info", {}).get("version"):
logger.warning(f"Specific version {version} requested but not implemented yet. "
f"Returning dependencies for latest version.")
return format_dependency_info(package_data)
except PyPIError:
# Re-raise PyPI-specific errors
raise
except Exception as e:
logger.error(f"Unexpected error querying dependencies for {package_name}: {e}")
raise NetworkError(f"Failed to query package dependencies: {e}", e) from e