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

Merge fix/version-parameter: Enable version parameter functionality

Browse source
This commit is contained in:
Ryan Malloy 2025-08-15 11:54:20 -06:00
commit 9cc4798d1d
8 changed files with 1164 additions and 23 deletions
Download patch file Download diff file Expand all files Collapse all files

36
pypi_query_mcp/core/pypi_client.py
View file

@ -164,12 +164,13 @@ class PyPIClient:
raise last_exception
async def get_package_info(
self, package_name: str, use_cache: bool = True
self, package_name: str, version: str | None = None, use_cache: bool = True
) -> dict[str, Any]:
"""Get comprehensive package information from PyPI.
Args:
package_name: Name of the package to query
version: Specific version to query (optional, defaults to latest)
use_cache: Whether to use cached data if available
Returns:
@ -177,22 +178,29 @@ class PyPIClient:
Raises:
InvalidPackageNameError: If package name is invalid
PackageNotFoundError: If package is not found
PackageNotFoundError: If package is not found or version doesn't exist
NetworkError: For network-related errors
"""
normalized_name = self._validate_package_name(package_name)
cache_key = self._get_cache_key(normalized_name, "info")
# Create cache key that includes version info
cache_suffix = f"v{version}" if version else "latest"
cache_key = self._get_cache_key(normalized_name, f"info_{cache_suffix}")
# Check cache first
if use_cache and cache_key in self._cache:
cache_entry = self._cache[cache_key]
if self._is_cache_valid(cache_entry):
logger.debug(f"Using cached data for package: {normalized_name}")
logger.debug(f"Using cached data for package: {normalized_name} version: {version or 'latest'}")
return cache_entry["data"]
# Make API request
url = f"{self.base_url}/{quote(normalized_name)}/json"
logger.info(f"Fetching package info for: {normalized_name}")
# Build URL - include version if specified
if version:
url = f"{self.base_url}/{quote(normalized_name)}/{quote(version)}/json"
logger.info(f"Fetching package info for: {normalized_name} version {version}")
else:
url = f"{self.base_url}/{quote(normalized_name)}/json"
logger.info(f"Fetching package info for: {normalized_name} (latest)")
try:
data = await self._make_request(url)
@ -204,8 +212,16 @@ class PyPIClient:
return data
except PackageNotFoundError as e:
if version:
# More specific error message for version not found
logger.error(f"Version {version} not found for package {normalized_name}")
raise PackageNotFoundError(f"Version {version} not found for package {normalized_name}")
else:
logger.error(f"Failed to fetch package info for {normalized_name}: {e}")
raise
except Exception as e:
logger.error(f"Failed to fetch package info for {normalized_name}: {e}")
logger.error(f"Failed to fetch package info for {normalized_name} version {version or 'latest'}: {e}")
raise
async def get_package_versions(
@ -220,7 +236,7 @@ class PyPIClient:
Returns:
List of version strings
"""
package_info = await self.get_package_info(package_name, use_cache)
package_info = await self.get_package_info(package_name, version=None, use_cache=use_cache)
releases = package_info.get("releases", {})
return list(releases.keys())
@ -236,7 +252,7 @@ class PyPIClient:
Returns:
Latest version string
"""
package_info = await self.get_package_info(package_name, use_cache)
package_info = await self.get_package_info(package_name, version=None, use_cache=use_cache)
return package_info.get("info", {}).get("version", "")
def clear_cache(self):

41
pypi_query_mcp/tools/package_query.py
View file

@ -1,6 +1,7 @@
"""Package query tools for PyPI MCP server."""
import logging
import re
from typing import Any
from ..core import InvalidPackageNameError, NetworkError, PyPIClient, PyPIError
@ -9,6 +10,24 @@ from ..core.version_utils import sort_versions_semantically
logger = logging.getLogger(__name__)
def validate_version_format(version: str | None) -> bool:
"""Validate that a version string follows a reasonable format.
Args:
version: Version string to validate
Returns:
True if version format is valid or None, False otherwise
"""
if version is None:
return True
# Basic validation for common version patterns
# Supports: 1.0.0, 1.0, 1.0.0a1, 1.0.0b2, 1.0.0rc1, 1.0.0.dev1, 2.0.0-dev, etc.
version_pattern = r"^[0-9]+(?:\.[0-9]+)*(?:[\.\-]?(?:a|b|rc|alpha|beta|dev|pre|post|final)[0-9]*)*$"
return bool(re.match(version_pattern, version.strip(), re.IGNORECASE))
def format_package_info(package_data: dict[str, Any]) -> dict[str, Any]:
"""Format package information for MCP response.
@ -186,7 +205,7 @@ async def query_package_info(package_name: str) -> dict[str, Any]:
try:
async with PyPIClient() as client:
package_data = await client.get_package_info(package_name)
package_data = await client.get_package_info(package_name, version=None)
return format_package_info(package_data)
except PyPIError:
# Re-raise PyPI-specific errors
@ -217,7 +236,7 @@ async def query_package_versions(package_name: str) -> dict[str, Any]:
try:
async with PyPIClient() as client:
package_data = await client.get_package_info(package_name)
package_data = await client.get_package_info(package_name, version=None)
return format_version_info(package_data)
except PyPIError:
# Re-raise PyPI-specific errors
@ -241,12 +260,16 @@ async def query_package_dependencies(
Raises:
InvalidPackageNameError: If package name is invalid
PackageNotFoundError: If package is not found
PackageNotFoundError: If package is not found or version doesn't exist
NetworkError: For network-related errors
"""
if not package_name or not package_name.strip():
raise InvalidPackageNameError(package_name)
# Validate version format if provided
if version and not validate_version_format(version):
raise InvalidPackageNameError(f"Invalid version format: {version}")
logger.info(
f"Querying dependencies for package: {package_name}"
+ (f" version {version}" if version else " (latest)")
@ -254,16 +277,8 @@ async def query_package_dependencies(
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."
)
# Pass the version parameter to get_package_info
package_data = await client.get_package_info(package_name, version=version)
return format_dependency_info(package_data)
except PyPIError:
# Re-raise PyPI-specific errors