Mcp Builder

"""
FastMCP API Client Pattern
===========================
Manual API integration with connection pooling, caching, and retry logic.
"""

from fastmcp import FastMCP
import httpx
import os
import time
import asyncio
from typing import Optional, Any, Dict
from datetime import datetime
from dotenv import load_dotenv

load_dotenv()

mcp = FastMCP("API Client Pattern")

# ============================================================================
# Configuration
# ============================================================================

class Config:
    """API configuration from environment."""
    API_BASE_URL = os.getenv("API_BASE_URL", "https://api.example.com")
    API_KEY = os.getenv("API_KEY", "")
    API_TIMEOUT = int(os.getenv("API_TIMEOUT", "30"))
    CACHE_TTL = int(os.getenv("CACHE_TTL", "300"))  # 5 minutes
    MAX_RETRIES = int(os.getenv("MAX_RETRIES", "3"))


# ============================================================================
# API Client with Connection Pooling
# ============================================================================

class APIClient:
    """Singleton API client with connection pooling."""
    _instance: Optional[httpx.AsyncClient] = None

    @classmethod
    async def get_client(cls) -> httpx.AsyncClient:
        """Get or create the shared HTTP client."""
        if cls._instance is None:
            cls._instance = httpx.AsyncClient(
                base_url=Config.API_BASE_URL,
                headers={
                    "Authorization": f"Bearer {Config.API_KEY}",
                    "Content-Type": "application/json",
                    "User-Agent": "FastMCP-Client/1.0"
                },
                timeout=httpx.Timeout(Config.API_TIMEOUT),
                limits=httpx.Limits(
                    max_keepalive_connections=5,
                    max_connections=10
                )
            )
        return cls._instance

    @classmethod
    async def cleanup(cls):
        """Cleanup the HTTP client."""
        if cls._instance:
            await cls._instance.aclose()
            cls._instance = None


# ============================================================================
# Cache Implementation
# ============================================================================

class SimpleCache:
    """Time-based cache for API responses."""

    def __init__(self, ttl: int = 300):
        self.ttl = ttl
        self.cache: Dict[str, Any] = {}
        self.timestamps: Dict[str, float] = {}

    def get(self, key: str) -> Optional[Any]:
        """Get cached value if not expired."""
        if key in self.cache:
            if time.time() - self.timestamps[key] < self.ttl:
                return self.cache[key]
            else:
                # Expired, remove it
                del self.cache[key]
                del self.timestamps[key]
        return None

    def set(self, key: str, value: Any):
        """Set cache value with timestamp."""
        self.cache[key] = value
        self.timestamps[key] = time.time()

    def invalidate(self, pattern: Optional[str] = None):
        """Invalidate cache entries."""
        if pattern:
            keys_to_delete = [k for k in self.cache if pattern in k]
            for key in keys_to_delete:
                del self.cache[key]
                del self.timestamps[key]
        else:
            self.cache.clear()
            self.timestamps.clear()


# Global cache instance
cache = SimpleCache(ttl=Config.CACHE_TTL)


# ============================================================================
# Retry Logic with Exponential Backoff
# ============================================================================

async def retry_with_backoff(
    func,
    max_retries: int = 3,
    initial_delay: float = 1.0,
    exponential_base: float = 2.0
):
    """Retry function with exponential backoff."""
    delay = initial_delay
    last_exception = None

    for attempt in range(max_retries):
        try: