src/mcp_registry.py

"""MCP Registry - Track generated MCPs to avoid duplicates"""

import json
from pathlib import Path
from typing import Optional, Dict, List
from datetime import datetime
import hashlib

from .config import HOSTED_MCPS_DIR


class MCPRegistry:
    """Registry for tracking generated MCPs"""

    def __init__(self):
        self.registry_file = HOSTED_MCPS_DIR / "registry.json"
        self.registry: Dict[str, dict] = {}
        self._load()

    def _load(self):
        """Load registry from disk"""
        if self.registry_file.exists():
            try:
                with open(self.registry_file, 'r') as f:
                    self.registry = json.load(f)
            except Exception:
                self.registry = {}
        else:
            self.registry = {}

    def _save(self):
        """Save registry to disk"""
        self.registry_file.parent.mkdir(exist_ok=True)
        with open(self.registry_file, 'w') as f:
            json.dump(self.registry, indent=2, fp=f)

    def _url_hash(self, api_url: str) -> str:
        """Generate a consistent hash for an API URL"""
        # Normalize URL (remove trailing slashes, lowercase)
        normalized = api_url.lower().rstrip('/')
        return hashlib.md5(normalized.encode()).hexdigest()

    def find_by_url(self, api_url: str) -> Optional[dict]:
        """Find an existing MCP for a given API URL

        Args:
            api_url: The API URL to search for

        Returns:
            MCP info dict if found, None otherwise
        """
        url_hash = self._url_hash(api_url)

        if url_hash in self.registry:
            mcp_info = self.registry[url_hash]

            # Check if the MCP directory still exists
            mcp_path = HOSTED_MCPS_DIR / mcp_info['mcp_id']
            if mcp_path.exists():
                return mcp_info
            else:
                # Clean up stale entry
                del self.registry[url_hash]
                self._save()

        return None

    def register(self, api_url: str, mcp_id: str, api_name: str, metadata: dict = None) -> dict:
        """Register a new MCP

        Args:
            api_url: The API URL
            mcp_id: The generated MCP ID
            api_name: The API name
            metadata: Optional metadata dict

        Returns:
            The registered MCP info
        """
        url_hash = self._url_hash(api_url)

        mcp_info = {
            'mcp_id': mcp_id,
            'api_url': api_url,
            'api_name': api_name,
            'created_at': datetime.now().isoformat(),
            'last_used': datetime.now().isoformat(),
            'metadata': metadata or {}
        }

        self.registry[url_hash] = mcp_info
        self._save()

        return mcp_info

    def update_last_used(self, api_url: str):
        """Update the last_used timestamp for an MCP

        Args:
            api_url: The API URL
        """
        url_hash = self._url_hash(api_url)

        if url_hash in self.registry:
            self.registry[url_hash]['last_used'] = datetime.now().isoformat()
            self._save()

    def list_all(self) -> List[dict]:
        """List all registered MCPs

        Returns:
            List of MCP info dicts
        """
        # Filter out MCPs whose directories no longer exist
        valid_mcps = []

        for url_hash, mcp_info in list(self.registry.items()):
            mcp_path = HOSTED_MCPS_DIR / mcp_info['mcp_id']
            if mcp_path.exists():
                valid_mcps.append(mcp_info)
            else:
                # Clean up stale entry
                del self.registry[url_hash]

        if len(valid_mcps) != len(self.registry):
            self._save()

        # Sort by last_used, most recent first
        valid_mcps.sort(key=lambda x: x['last_used'], reverse=True)

        return valid_mcps

    def delete(self, api_url: str) -> bool:
        """Delete an MCP from registry

        Args:
            api_url: The API URL

        Returns:
            True if deleted, False if not found
        """
        url_hash = self._url_hash(api_url)

        if url_hash in self.registry:
            del self.registry[url_hash]
            self._save()
            return True

        return False

    def get_stats(self) -> dict:
        """Get registry statistics

        Returns:
            Stats dict
        """
        mcps = self.list_all()

        return {
            'total_mcps': len(mcps),
            'oldest': mcps[-1]['created_at'] if mcps else None,
            'newest': mcps[0]['created_at'] if mcps else None,
        }


# Global registry instance
mcp_registry = MCPRegistry()