"""Package management using uv for MCP Python REPL server."""
import logging
import subprocess
import json
from pathlib import Path
from typing import List, Dict, Any
from .models import PackageInfo
logger = logging.getLogger(__name__)
class PackageManager:
"""Manages Python packages using uv."""
def __init__(self, project_dir: Path | None = None):
"""Initialize PackageManager with project directory.
Args:
project_dir: Project directory path. If None, uses current working directory.
"""
self.project_dir = project_dir or Path.cwd()
async def install_packages(self, packages: List[str], dev: bool = False) -> Dict[str, Any]:
"""Install Python packages using uv add.
Args:
packages: List of package names to install
dev: Whether to install as development dependencies
Returns:
Dictionary with installation result
"""
try:
cmd = ["uv", "add"]
if dev:
cmd.append("--dev")
cmd.extend(packages)
logger.info(f"Installing packages: {packages} (dev={dev})")
# subprocess.run is synchronous - do NOT await it
result = subprocess.run(
cmd,
cwd=self.project_dir,
capture_output=True,
text=True
)
if result.returncode == 0:
return {
"success": True,
"message": f"Successfully installed: {', '.join(packages)}",
"output": result.stdout.strip(),
"packages": packages,
"dev": dev
}
else:
return {
"success": False,
"error": result.stderr.strip(),
"message": f"Failed to install packages: {', '.join(packages)}",
"returncode": result.returncode
}
except Exception as e:
logger.error(f"Package installation error: {e}")
return {
"success": False,
"error": str(e),
"message": "Package installation failed due to unexpected error"
}
async def remove_packages(self, packages: List[str]) -> Dict[str, Any]:
"""Remove Python packages using uv remove.
Args:
packages: List of package names to remove
Returns:
Dictionary with removal result
"""
try:
cmd = ["uv", "remove"] + packages
logger.info(f"Removing packages: {packages}")
# subprocess.run is synchronous - do NOT await it
result = subprocess.run(
cmd,
cwd=self.project_dir,
capture_output=True,
text=True
)
if result.returncode == 0:
return {
"success": True,
"message": f"Successfully removed: {', '.join(packages)}",
"output": result.stdout.strip(),
"packages": packages
}
else:
return {
"success": False,
"error": result.stderr.strip(),
"message": f"Failed to remove packages: {', '.join(packages)}",
"returncode": result.returncode
}
except Exception as e:
logger.error(f"Package removal error: {e}")
return {
"success": False,
"error": str(e),
"message": "Package removal failed due to unexpected error"
}
async def list_packages(self) -> List[PackageInfo]:
"""List installed packages using uv pip list.
Returns:
List of PackageInfo objects
"""
try:
# Use uv pip list to get package information
cmd = ["uv", "pip", "list", "--format", "json"]
logger.info("Listing installed packages")
# subprocess.run is synchronous - do NOT await it
result = subprocess.run(
cmd,
cwd=self.project_dir,
capture_output=True,
text=True
)
if result.returncode == 0:
try:
packages_data = json.loads(result.stdout)
packages = []
for pkg_data in packages_data:
package = PackageInfo(
name=pkg_data.get("name", ""),
version=pkg_data.get("version", ""),
dev_dependency=False # uv pip list doesn't distinguish dev deps
)
packages.append(package)
return packages
except json.JSONDecodeError as e:
logger.error(f"Failed to parse package list JSON: {e}")
return []
else:
logger.error(f"Package list command failed: {result.stderr}")
return []
except Exception as e:
logger.error(f"Package listing error: {e}")
return []
def is_uv_available(self) -> bool:
"""Check if uv command is available.
Returns:
True if uv is available, False otherwise
"""
try:
result = subprocess.run(
["uv", "--version"],
capture_output=True,
text=True
)
return result.returncode == 0
except (subprocess.SubprocessError, FileNotFoundError):
return False
