Production Deployment

Comprehensive guide to deploying, scaling, and operating MCP-enabled agents in production environments with best practices for reliability, security, and performance.

Overview

Production deployment of MCP-enabled agents requires careful consideration of reliability, security, performance, and operational requirements. This guide provides battle-tested patterns for production success.

Production vs. Development

# ❌ Development configuration - Not production ready
dev_team = (MCPTeamBuilder("dev_team")
    .with_llm(ChatOpenAI(model="gpt-4o-mini"))
    .with_mcp_server("python", ["server.py"])
    .build()
)

# ✅ Production configuration - Hardened and monitored
prod_team = (MCPTeamBuilder("prod_team")
    .with_llm(ChatOpenAI(
        model="gpt-4o-mini",
        temperature=0.7,
        timeout=30,
        max_retries=3,
        request_timeout=60
    ))
    .with_mcp_server(
        "python",
        ["/opt/mcp/servers/server.py"],
        env={
            "LOG_LEVEL": "INFO",
            "MAX_RETRIES": "3",
            "TIMEOUT": "30",
            "ENVIRONMENT": "production"
        },
        timeout=30
    )
    .skip_failed_servers(True)  # Graceful degradation
    .build()
)

Key Production Requirements

1. Reliability: 99.9%+ uptime, fault tolerance
2. Security: Authentication, encryption, audit logs
3. Performance: Low latency, high throughput
4. Monitoring: Metrics, logs, alerts
5. Scalability: Handle increasing load
6. Maintainability: Easy updates and debugging

Pre-Production Checklist

Infrastructure Checklist

# production_checklist.yaml
infrastructure:
  compute:
    - [ ] Production servers provisioned
    - [ ] Resource limits configured (CPU, memory)
    - [ ] Auto-scaling configured
    - [ ] Load balancer set up

  networking:
    - [ ] VPC/network configured
    - [ ] Firewall rules configured
    - [ ] SSL/TLS certificates installed
    - [ ] DNS configured

  storage:
    - [ ] Database provisioned and backed up
    - [ ] File storage configured
    - [ ] Backup strategy implemented
    - [ ] Data retention policies defined

security:
  - [ ] Secrets management configured
  - [ ] API keys rotated
  - [ ] SSL/TLS enabled
  - [ ] Rate limiting configured
  - [ ] WAF/DDoS protection enabled
  - [ ] Security audit completed
  - [ ] Compliance requirements met

monitoring:
  - [ ] Logging infrastructure set up
  - [ ] Metrics collection configured
  - [ ] Alerting rules defined
  - [ ] Dashboard created
  - [ ] On-call rotation established
  - [ ] Runbook created

deployment:
  - [ ] CI/CD pipeline configured
  - [ ] Blue-green/canary deployment ready
  - [ ] Rollback procedure tested
  - [ ] Health checks implemented
  - [ ] Load testing completed
  - [ ] Disaster recovery tested

documentation:
  - [ ] Architecture documented
  - [ ] API documentation complete
  - [ ] Operational runbooks created
  - [ ] Incident response plan defined
  - [ ] SLA defined and communicated

Code Quality Checklist

"""
Pre-production code quality checklist.
"""

def production_readiness_check():
    """Verify production readiness."""

    checks = {
        "error_handling": False,
        "logging": False,
        "monitoring": False,
        "configuration": False,
        "security": False,
        "testing": False,
        "documentation": False
    }

    # 1. Error handling
    try:
        # Check comprehensive error handling
        checks["error_handling"] = verify_error_handling()
    except:
        pass

    # 2. Logging
    try:
        # Check structured logging
        checks["logging"] = verify_logging_setup()
    except:
        pass

    # 3. Monitoring
    try:
        # Check metrics and health endpoints
        checks["monitoring"] = verify_monitoring()
    except:
        pass

    # 4. Configuration
    try:
        # Check environment-based config
        checks["configuration"] = verify_configuration()
    except:
        pass

    # 5. Security
    try:
        # Check secrets management
        checks["security"] = verify_security()
    except:
        pass

    # 6. Testing
    try:
        # Check test coverage
        checks["testing"] = verify_test_coverage()
    except:
        pass

    # 7. Documentation
    try:
        # Check documentation completeness
        checks["documentation"] = verify_documentation()
    except:
        pass

    # Report results
    passed = sum(checks.values())
    total = len(checks)

    print(f"Production Readiness: {passed}/{total} checks passed")
    for check, status in checks.items():
        status_str = "✅" if status else "❌"
        print(f"{status_str} {check}")

    return all(checks.values())

Infrastructure Setup

Docker Deployment

# Dockerfile
FROM python:3.11-slim

# Set working directory
WORKDIR /app

# Install system dependencies
RUN apt-get update && apt-get install -y \
    build-essential \
    curl \
    git \
    && rm -rf /var/lib/apt/lists/*

# Install Node.js (for npx-based MCP servers)
RUN curl -fsSL https://deb.nodesource.com/setup_18.x | bash - \
    && apt-get install -y nodejs \
    && rm -rf /var/lib/apt/lists/*

# Copy requirements
COPY requirements.txt .

# Install Python dependencies
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY . .

# Create necessary directories
RUN mkdir -p /app/logs /app/data /app/rl_data

# Set environment variables
ENV PYTHONUNBUFFERED=1
ENV LOG_LEVEL=INFO
ENV ENVIRONMENT=production

# Expose port (if using HTTP/SSE)
EXPOSE 8000

# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=40s --retries=3 \
    CMD python -c "import requests; requests.get('http://localhost:8000/health')"

# Run application
CMD ["python", "main.py"]

# docker-compose.yml
version: '3.8'

services:
  mcp-agent:
    build: .
    container_name: mcp-agent-prod
    restart: unless-stopped

    ports:
      - "8000:8000"

    environment:
      - LOG_LEVEL=INFO
      - ENVIRONMENT=production
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - GITHUB_TOKEN=${GITHUB_TOKEN}
      - DATABASE_URL=${DATABASE_URL}

    volumes:
      - ./logs:/app/logs
      - ./data:/app/data
      - ./rl_data:/app/rl_data
      - ./config:/app/config:ro

    networks:
      - mcp-network

    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 40s

    logging:
      driver: "json-file"
      options:
        max-size: "10m"
        max-file: "3"

    deploy:
      resources:
        limits:
          cpus: '2'
          memory: 4G
        reservations:
          cpus: '1'
          memory: 2G

  # Redis for caching (optional)
  redis:
    image: redis:7-alpine
    container_name: mcp-redis
    restart: unless-stopped

    networks:
      - mcp-network

    volumes:
      - redis-data:/data

    command: redis-server --appendonly yes

  # PostgreSQL for persistence (optional)
  postgres:
    image: postgres:15-alpine
    container_name: mcp-postgres
    restart: unless-stopped

    environment:
      - POSTGRES_DB=mcp_db
      - POSTGRES_USER=mcp_user
      - POSTGRES_PASSWORD=${POSTGRES_PASSWORD}

    networks:
      - mcp-network

    volumes:
      - postgres-data:/var/lib/postgresql/data

    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U mcp_user"]
      interval: 10s
      timeout: 5s
      retries: 5

networks:
  mcp-network:
    driver: bridge

volumes:
  redis-data:
  postgres-data:

Kubernetes Deployment

# kubernetes/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: mcp-agent
  namespace: production
  labels:
    app: mcp-agent
    version: v1.0.0
spec:
  replicas: 3
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0
  selector:
    matchLabels:
      app: mcp-agent
  template:
    metadata:
      labels:
        app: mcp-agent
        version: v1.0.0
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "8000"
        prometheus.io/path: "/metrics"
    spec:
      containers:
      - name: mcp-agent
        image: myregistry/mcp-agent:v1.0.0
        imagePullPolicy: Always

        ports:
        - containerPort: 8000
          name: http
          protocol: TCP

        env:
        - name: LOG_LEVEL
          value: "INFO"
        - name: ENVIRONMENT
          value: "production"
        - name: OPENAI_API_KEY
          valueFrom:
            secretKeyRef:
              name: mcp-secrets
              key: openai-api-key
        - name: GITHUB_TOKEN
          valueFrom:
            secretKeyRef:
              name: mcp-secrets
              key: github-token

        resources:
          requests:
            cpu: "500m"
            memory: "1Gi"
          limits:
            cpu: "2000m"
            memory: "4Gi"

        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
          timeoutSeconds: 5
          failureThreshold: 3

        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 10
          periodSeconds: 5
          timeoutSeconds: 3
          failureThreshold: 3

        volumeMounts:
        - name: config
          mountPath: /app/config
          readOnly: true
        - name: data
          mountPath: /app/data
        - name: logs
          mountPath: /app/logs

      volumes:
      - name: config
        configMap:
          name: mcp-config
      - name: data
        persistentVolumeClaim:
          claimName: mcp-data-pvc
      - name: logs
        emptyDir: {}

      securityContext:
        runAsNonRoot: true
        runAsUser: 1000
        fsGroup: 1000

---
# kubernetes/service.yaml
apiVersion: v1
kind: Service
metadata:
  name: mcp-agent-service
  namespace: production
  labels:
    app: mcp-agent
spec:
  type: ClusterIP
  ports:
  - port: 80
    targetPort: 8000
    protocol: TCP
    name: http
  selector:
    app: mcp-agent

---
# kubernetes/hpa.yaml
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: mcp-agent-hpa
  namespace: production
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: mcp-agent
  minReplicas: 3
  maxReplicas: 10
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80

Configuration Management

Environment-Based Configuration

"""
production_config.py - Production configuration management.
"""

import os
from enum import Enum
from typing import Dict, Any, Optional
from pydantic import BaseSettings, Field, validator
import logging

logger = logging.getLogger(__name__)


class Environment(str, Enum):
    """Deployment environment."""
    DEVELOPMENT = "development"
    STAGING = "staging"
    PRODUCTION = "production"


class MCPServerConfig(BaseSettings):
    """Configuration for MCP server."""
    name: str
    command: Optional[str] = None
    args: list = []
    url: Optional[str] = None
    transport: str = "stdio"
    timeout: int = 30
    env_vars: Dict[str, str] = {}
    optional: bool = False

    class Config:
        env_prefix = "MCP_SERVER_"


class ProductionConfig(BaseSettings):
    """Production configuration."""

    # Environment
    environment: Environment = Field(default=Environment.PRODUCTION)
    debug: bool = Field(default=False)

    # LLM Configuration
    llm_model: str = Field(default="gpt-4o-mini")
    llm_temperature: float = Field(default=0.7)
    llm_max_tokens: int = Field(default=2000)
    llm_timeout: int = Field(default=60)
    llm_max_retries: int = Field(default=3)

    # API Keys (from environment)
    openai_api_key: str = Field(..., env="OPENAI_API_KEY")
    github_token: Optional[str] = Field(None, env="GITHUB_TOKEN")

    # Database
    database_url: Optional[str] = Field(None, env="DATABASE_URL")
    database_pool_size: int = Field(default=10)
    database_max_overflow: int = Field(default=20)

    # Redis
    redis_url: Optional[str] = Field(None, env="REDIS_URL")
    redis_ttl: int = Field(default=3600)

    # Logging
    log_level: str = Field(default="INFO")
    log_format: str = Field(default="json")
    log_file: Optional[str] = Field(default="/app/logs/mcp.log")

    # Monitoring
    enable_metrics: bool = Field(default=True)
    metrics_port: int = Field(default=9090)
    enable_tracing: bool = Field(default=True)
    tracing_endpoint: Optional[str] = Field(None, env="TRACING_ENDPOINT")

    # Performance
    max_concurrent_requests: int = Field(default=100)
    request_timeout: int = Field(default=300)
    enable_caching: bool = Field(default=True)
    cache_ttl: int = Field(default=3600)

    # Security
    enable_auth: bool = Field(default=True)
    jwt_secret: Optional[str] = Field(None, env="JWT_SECRET")
    allowed_origins: list = Field(default=["*"])
    rate_limit_per_minute: int = Field(default=60)

    # MCP Servers
    mcp_servers: list = Field(default_factory=list)

    # RL Configuration
    enable_rl: bool = Field(default=True)
    rl_exploration_rate: float = Field(default=0.15)
    rl_learning_rate: float = Field(default=0.1)
    rl_q_table_path: str = Field(default="/app/rl_data/q_table.pkl")

    # Health Check
    health_check_interval: int = Field(default=30)

    class Config:
        env_file = ".env.production"
        env_file_encoding = "utf-8"

    @validator("environment", pre=True)
    def validate_environment(cls, v):
        """Validate environment."""
        if isinstance(v, str):
            return Environment(v.lower())
        return v

    @validator("log_level")
    def validate_log_level(cls, v):
        """Validate log level."""
        valid_levels = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
        if v.upper() not in valid_levels:
            raise ValueError(f"Invalid log level. Must be one of {valid_levels}")
        return v.upper()

    def get_llm_config(self) -> Dict[str, Any]:
        """Get LLM configuration."""
        return {
            "model": self.llm_model,
            "temperature": self.llm_temperature,
            "max_tokens": self.llm_max_tokens,
            "timeout": self.llm_timeout,
            "max_retries": self.llm_max_retries,
            "api_key": self.openai_api_key
        }

    def get_database_url(self) -> str:
        """Get database connection URL."""
        if not self.database_url:
            raise ValueError("DATABASE_URL not configured")
        return self.database_url

    def is_production(self) -> bool:
        """Check if running in production."""
        return self.environment == Environment.PRODUCTION

    def get_log_config(self) -> Dict[str, Any]:
        """Get logging configuration."""
        return {
            "level": self.log_level,
            "format": self.log_format,
            "file": self.log_file
        }


def load_config() -> ProductionConfig:
    """Load production configuration."""
    try:
        config = ProductionConfig()
        logger.info(f"Loaded configuration for environment: {config.environment.value}")
        return config
    except Exception as e:
        logger.error(f"Failed to load configuration: {e}")
        raise


# Global configuration instance
config = load_config()

Secrets Management

"""
secrets_manager.py - Secure secrets management.
"""

import os
import boto3
from typing import Dict, Any, Optional
import logging

logger = logging.getLogger(__name__)


class SecretsManager:
    """Manage secrets securely."""

    def __init__(self, provider: str = "env"):
        """
        Initialize secrets manager.

        Args:
            provider: Secrets provider ('env', 'aws', 'vault', 'k8s')
        """
        self.provider = provider
        self._cache: Dict[str, str] = {}

        if provider == "aws":
            self.client = boto3.client('secretsmanager')
        elif provider == "vault":
            # Initialize HashiCorp Vault client
            pass
        elif provider == "k8s":
            # Initialize Kubernetes secrets client
            pass

    def get_secret(self, key: str, required: bool = True) -> Optional[str]:
        """
        Get secret value.

        Args:
            key: Secret key
            required: Whether secret is required

        Returns:
            Secret value or None
        """
        # Check cache first
        if key in self._cache:
            return self._cache[key]

        # Fetch from provider
        value = None

        if self.provider == "env":
            value = os.getenv(key)

        elif self.provider == "aws":
            try:
                response = self.client.get_secret_value(SecretId=key)
                value = response['SecretString']
            except Exception as e:
                logger.error(f"Failed to fetch secret from AWS: {e}")

        elif self.provider == "vault":
            # Fetch from Vault
            pass

        elif self.provider == "k8s":
            # Fetch from Kubernetes secrets
            pass

        # Validate
        if required and not value:
            raise ValueError(f"Required secret not found: {key}")

        # Cache
        if value:
            self._cache[key] = value

        return value

    def get_database_credentials(self) -> Dict[str, str]:
        """Get database credentials."""
        return {
            "host": self.get_secret("DB_HOST"),
            "port": self.get_secret("DB_PORT"),
            "database": self.get_secret("DB_NAME"),
            "user": self.get_secret("DB_USER"),
            "password": self.get_secret("DB_PASSWORD")
        }

    def get_api_keys(self) -> Dict[str, str]:
        """Get API keys."""
        return {
            "openai": self.get_secret("OPENAI_API_KEY"),
            "github": self.get_secret("GITHUB_TOKEN", required=False),
            "slack": self.get_secret("SLACK_BOT_TOKEN", required=False)
        }

    def rotate_secret(self, key: str, new_value: str):
        """Rotate a secret."""
        if self.provider == "aws":
            try:
                self.client.update_secret(SecretId=key, SecretString=new_value)
                logger.info(f"Rotated secret: {key}")
                # Clear cache
                if key in self._cache:
                    del self._cache[key]
            except Exception as e:
                logger.error(f"Failed to rotate secret: {e}")
                raise

    def clear_cache(self):
        """Clear secrets cache."""
        self._cache.clear()
        logger.info("Cleared secrets cache")


# Global secrets manager
secrets = SecretsManager(provider=os.getenv("SECRETS_PROVIDER", "env"))

Deployment Strategies

Blue-Green Deployment

"""
blue_green_deployment.py - Blue-green deployment strategy.
"""

import time
import logging
from enum import Enum
from typing import Optional

logger = logging.getLogger(__name__)


class DeploymentColor(Enum):
    """Deployment color."""
    BLUE = "blue"
    GREEN = "green"


class BlueGreenDeployment:
    """Manage blue-green deployments."""

    def __init__(self):
        self.active_color = DeploymentColor.BLUE
        self.blue_team = None
        self.green_team = None

    def deploy_new_version(self, new_team):
        """Deploy new version to inactive environment."""
        inactive_color = self._get_inactive_color()
        logger.info(f"Deploying new version to {inactive_color.value} environment")

        if inactive_color == DeploymentColor.BLUE:
            self.blue_team = new_team
        else:
            self.green_team = new_team

        logger.info(f"New version deployed to {inactive_color.value}")

    def health_check(self, team) -> bool:
        """Perform health check on team."""
        try:
            # Test basic functionality
            result = team({
                "messages": [HumanMessage(content="health check")]
            })
            return True
        except Exception as e:
            logger.error(f"Health check failed: {e}")
            return False

    def switch_traffic(self):
        """Switch traffic to new version."""
        inactive_color = self._get_inactive_color()
        inactive_team = self._get_team(inactive_color)

        if not inactive_team:
            raise ValueError(f"No team deployed to {inactive_color.value}")

        # Health check before switching
        logger.info(f"Performing health check on {inactive_color.value}")
        if not self.health_check(inactive_team):
            raise RuntimeError(f"Health check failed for {inactive_color.value}")

        # Switch traffic
        logger.info(f"Switching traffic from {self.active_color.value} to {inactive_color.value}")
        self.active_color = inactive_color
        logger.info(f"Traffic switched to {self.active_color.value}")

    def rollback(self):
        """Rollback to previous version."""
        previous_color = self._get_inactive_color()
        logger.warning(f"Rolling back to {previous_color.value}")
        self.active_color = previous_color
        logger.info(f"Rolled back to {self.active_color.value}")

    def get_active_team(self):
        """Get currently active team."""
        return self._get_team(self.active_color)

    def _get_inactive_color(self) -> DeploymentColor:
        """Get inactive color."""
        return (
            DeploymentColor.GREEN
            if self.active_color == DeploymentColor.BLUE
            else DeploymentColor.BLUE
        )

    def _get_team(self, color: DeploymentColor):
        """Get team by color."""
        return self.blue_team if color == DeploymentColor.BLUE else self.green_team


# Usage example
def perform_blue_green_deployment():
    """Perform blue-green deployment."""
    deployment = BlueGreenDeployment()

    # Current active version (blue)
    logger.info("Current version running on blue")

    # Deploy new version to green
    logger.info("Building new version...")
    new_team = build_new_team_version()

    deployment.deploy_new_version(new_team)

    # Run smoke tests
    logger.info("Running smoke tests...")
    time.sleep(5)

    # Switch traffic
    try:
        deployment.switch_traffic()
        logger.info("Deployment successful!")
    except Exception as e:
        logger.error(f"Deployment failed: {e}")
        deployment.rollback()
        logger.info("Rolled back to previous version")

Canary Deployment

"""
canary_deployment.py - Canary deployment strategy.
"""

import random
import logging
from typing import List, Dict, Any

logger = logging.getLogger(__name__)


class CanaryDeployment:
    """Manage canary deployments."""

    def __init__(
        self,
        stable_team,
        canary_percentage: int = 10
    ):
        self.stable_team = stable_team
        self.canary_team = None
        self.canary_percentage = canary_percentage
        self.metrics: Dict[str, List[float]] = {
            "stable": [],
            "canary": []
        }

    def deploy_canary(self, canary_team):
        """Deploy canary version."""
        logger.info(f"Deploying canary with {self.canary_percentage}% traffic")
        self.canary_team = canary_team

    def route_request(self, request: Dict[str, Any]):
        """Route request to stable or canary."""
        # Decide which version to use
        use_canary = (
            self.canary_team is not None and
            random.random() * 100 < self.canary_percentage
        )

        team = self.canary_team if use_canary else self.stable_team
        version = "canary" if use_canary else "stable"

        logger.info(f"Routing to {version} version")

        # Execute request
        try:
            start_time = time.time()
            result = team(request)
            duration = time.time() - start_time

            # Record metrics
            self.metrics[version].append(duration)

            return {
                "result": result,
                "version": version,
                "duration": duration
            }

        except Exception as e:
            logger.error(f"Request failed on {version}: {e}")
            raise

    def increase_canary_traffic(self, increment: int = 10):
        """Gradually increase canary traffic."""
        if not self.canary_team:
            raise ValueError("No canary deployed")

        self.canary_percentage = min(100, self.canary_percentage + increment)
        logger.info(f"Increased canary traffic to {self.canary_percentage}%")

    def analyze_metrics(self) -> Dict[str, Any]:
        """Analyze performance metrics."""
        if not self.metrics["stable"] or not self.metrics["canary"]:
            return {"status": "insufficient_data"}

        stable_avg = sum(self.metrics["stable"]) / len(self.metrics["stable"])
        canary_avg = sum(self.metrics["canary"]) / len(self.metrics["canary"])

        # Calculate error rates (simplified)
        stable_errors = 0  # Track separately
        canary_errors = 0  # Track separately

        analysis = {
            "stable_avg_latency": stable_avg,
            "canary_avg_latency": canary_avg,
            "latency_diff_percent": ((canary_avg - stable_avg) / stable_avg) * 100,
            "stable_error_rate": stable_errors,
            "canary_error_rate": canary_errors,
            "recommendation": "proceed"
        }

        # Decision logic
        if canary_avg > stable_avg * 1.2:  # 20% slower
            analysis["recommendation"] = "rollback"
            analysis["reason"] = "Canary latency too high"
        elif canary_errors > stable_errors * 1.5:
            analysis["recommendation"] = "rollback"
            analysis["reason"] = "Canary error rate too high"

        return analysis

    def promote_canary(self):
        """Promote canary to stable."""
        if not self.canary_team:
            raise ValueError("No canary to promote")

        logger.info("Promoting canary to stable")
        self.stable_team = self.canary_team
        self.canary_team = None
        self.canary_percentage = 0
        logger.info("Canary promoted successfully")

    def rollback_canary(self):
        """Rollback canary deployment."""
        logger.warning("Rolling back canary")
        self.canary_team = None
        self.canary_percentage = 0
        logger.info("Canary rolled back")


# Usage example
def perform_canary_deployment():
    """Perform canary deployment."""
    stable_team = build_current_team()
    canary = CanaryDeployment(stable_team, canary_percentage=10)

    # Deploy canary
    new_team = build_new_team_version()
    canary.deploy_canary(new_team)

    # Gradually increase traffic
    for stage in [10, 25, 50, 100]:
        logger.info(f"Setting canary traffic to {stage}%")
        canary.canary_percentage = stage

        # Monitor for 10 minutes
        time.sleep(600)

        # Analyze metrics
        analysis = canary.analyze_metrics()

        if analysis["recommendation"] == "rollback":
            logger.error(f"Rolling back: {analysis['reason']}")
            canary.rollback_canary()
            break

        if stage == 100:
            logger.info("Canary successful, promoting to stable")
            canary.promote_canary()

Monitoring and Observability

Structured Logging

"""
production_logging.py - Production logging setup.
"""

import logging
import json
import sys
from datetime import datetime
from typing import Dict, Any
from pythonjsonlogger import jsonlogger


class ProductionLogger:
    """Production-ready logging."""

    def __init__(
        self,
        name: str,
        level: str = "INFO",
        log_file: str = None
    ):
        self.logger = logging.getLogger(name)
        self.logger.setLevel(level)

        # JSON formatter
        formatter = jsonlogger.JsonFormatter(
            fmt='%(asctime)s %(name)s %(levelname)s %(message)s',
            json_encoder=json.JSONEncoder
        )

        # Console handler
        console_handler = logging.StreamHandler(sys.stdout)
        console_handler.setFormatter(formatter)
        self.logger.addHandler(console_handler)

        # File handler (if specified)
        if log_file:
            file_handler = logging.FileHandler(log_file)
            file_handler.setFormatter(formatter)
            self.logger.addHandler(file_handler)

    def log_request(
        self,
        request_id: str,
        method: str,
        path: str,
        **kwargs
    ):
        """Log incoming request."""
        self.logger.info(
            "Request received",
            extra={
                "request_id": request_id,
                "method": method,
                "path": path,
                **kwargs
            }
        )

    def log_response(
        self,
        request_id: str,
        status_code: int,
        duration_ms: float,
        **kwargs
    ):
        """Log response."""
        self.logger.info(
            "Response sent",
            extra={
                "request_id": request_id,
                "status_code": status_code,
                "duration_ms": duration_ms,
                **kwargs
            }
        )

    def log_error(
        self,
        request_id: str,
        error: Exception,
        **kwargs
    ):
        """Log error."""
        self.logger.error(
            "Error occurred",
            extra={
                "request_id": request_id,
                "error_type": type(error).__name__,
                "error_message": str(error),
                **kwargs
            },
            exc_info=True
        )

    def log_metric(
        self,
        metric_name: str,
        value: float,
        **kwargs
    ):
        """Log metric."""
        self.logger.info(
            "Metric recorded",
            extra={
                "metric_name": metric_name,
                "value": value,
                **kwargs
            }
        )


# Setup production logger
prod_logger = ProductionLogger(
    name="mcp_production",
    level="INFO",
    log_file="/app/logs/mcp.log"
)

Metrics Collection

"""
production_metrics.py - Production metrics collection.
"""

from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time
from functools import wraps

# Define metrics
request_count = Counter(
    'mcp_requests_total',
    'Total number of requests',
    ['method', 'endpoint', 'status']
)

request_duration = Histogram(
    'mcp_request_duration_seconds',
    'Request duration in seconds',
    ['method', 'endpoint']
)

active_requests = Gauge(
    'mcp_active_requests',
    'Number of active requests'
)

tool_execution_count = Counter(
    'mcp_tool_executions_total',
    'Total tool executions',
    ['tool_name', 'status']
)

tool_execution_duration = Histogram(
    'mcp_tool_execution_duration_seconds',
    'Tool execution duration',
    ['tool_name']
)

rl_q_table_size = Gauge(
    'mcp_rl_q_table_size',
    'Size of RL Q-table'
)

rl_exploration_rate = Gauge(
    'mcp_rl_exploration_rate',
    'Current RL exploration rate'
)


def track_request(method: str, endpoint: str):
    """Decorator to track requests."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            active_requests.inc()
            start_time = time.time()

            try:
                result = func(*args, **kwargs)
                status = "success"
                return result
            except Exception as e:
                status = "error"
                raise
            finally:
                duration = time.time() - start_time
                active_requests.dec()
                request_count.labels(method=method, endpoint=endpoint, status=status).inc()
                request_duration.labels(method=method, endpoint=endpoint).observe(duration)

        return wrapper
    return decorator


def track_tool_execution(tool_name: str):
    """Decorator to track tool execution."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            start_time = time.time()

            try:
                result = func(*args, **kwargs)
                status = "success"
                return result
            except Exception as e:
                status = "error"
                raise
            finally:
                duration = time.time() - start_time
                tool_execution_count.labels(tool_name=tool_name, status=status).inc()
                tool_execution_duration.labels(tool_name=tool_name).observe(duration)

        return wrapper
    return decorator


def update_rl_metrics(rl_manager):
    """Update RL metrics."""
    stats = rl_manager.get_statistics()
    rl_q_table_size.set(stats["total_states"])
    rl_exploration_rate.set(stats["exploration_rate"])


def start_metrics_server(port: int = 9090):
    """Start Prometheus metrics server."""
    start_http_server(port)
    logger.info(f"Metrics server started on port {port}")

Health Checks

"""
production_health.py - Production health checks.
"""

from fastapi import FastAPI, Response
from typing import Dict, Any
import time

app = FastAPI()


class HealthChecker:
    """Comprehensive health checker."""

    def __init__(self, team):
        self.team = team
        self.last_check = None
        self.health_status = "unknown"

    def check_components(self) -> Dict[str, Any]:
        """Check all system components."""
        checks = {}

        # 1. MCP Servers
        try:
            server_count = self.team.get_mcp_server_count()
            checks["mcp_servers"] = {
                "status": "healthy",
                "count": server_count
            }
        except Exception as e:
            checks["mcp_servers"] = {
                "status": "unhealthy",
                "error": str(e)
            }

        # 2. Tools
        try:
            tools = self.team.get_mcp_tool_names()
            checks["tools"] = {
                "status": "healthy",
                "count": len(tools)
            }
        except Exception as e:
            checks["tools"] = {
                "status": "unhealthy",
                "error": str(e)
            }

        # 3. Database (if configured)
        try:
            if config.database_url:
                # Test database connection
                checks["database"] = {
                    "status": "healthy"
                }
        except Exception as e:
            checks["database"] = {
                "status": "unhealthy",
                "error": str(e)
            }

        # 4. Redis (if configured)
        try:
            if config.redis_url:
                # Test Redis connection
                checks["redis"] = {
                    "status": "healthy"
                }
        except Exception as e:
            checks["redis"] = {
                "status": "unhealthy",
                "error": str(e)
            }

        # Determine overall health
        all_healthy = all(
            check.get("status") == "healthy"
            for check in checks.values()
        )

        self.health_status = "healthy" if all_healthy else "unhealthy"
        self.last_check = time.time()

        return {
            "status": self.health_status,
            "timestamp": self.last_check,
            "checks": checks
        }


health_checker = None  # Initialize after team creation


@app.get("/health")
async def health():
    """Health check endpoint."""
    if health_checker:
        result = health_checker.check_components()
        status_code = 200 if result["status"] == "healthy" else 503
        return Response(
            content=json.dumps(result),
            status_code=status_code,
            media_type="application/json"
        )
    return {"status": "starting"}


@app.get("/ready")
async def readiness():
    """Readiness check endpoint."""
    # Check if system is ready to serve traffic
    if health_checker and health_checker.health_status == "healthy":
        return {"status": "ready"}
    return Response(
        content=json.dumps({"status": "not_ready"}),
        status_code=503,
        media_type="application/json"
    )


@app.get("/metrics")
async def metrics():
    """Metrics endpoint for Prometheus."""
    from prometheus_client import generate_latest, CONTENT_TYPE_LATEST
    return Response(
        content=generate_latest(),
        media_type=CONTENT_TYPE_LATEST
    )

Scaling and Performance

Horizontal Scaling

"""
horizontal_scaling.py - Horizontal scaling implementation.
"""

from typing import List
import random

class LoadBalancer:
    """Simple round-robin load balancer."""

    def __init__(self, instances: List):
        self.instances = instances
        self.current_index = 0

    def get_instance(self):
        """Get next instance (round-robin)."""
        instance = self.instances[self.current_index]
        self.current_index = (self.current_index + 1) % len(self.instances)
        return instance

    def add_instance(self, instance):
        """Add new instance."""
        self.instances.append(instance)
        logger.info(f"Added instance. Total: {len(self.instances)}")

    def remove_instance(self, instance):
        """Remove instance."""
        self.instances.remove(instance)
        logger.info(f"Removed instance. Total: {len(self.instances)}")


# Create multiple instances
instances = [
    build_team_instance(f"instance_{i}")
    for i in range(3)
]

load_balancer = LoadBalancer(instances)

# Route requests
def handle_request(request):
    """Handle request with load balancing."""
    instance = load_balancer.get_instance()
    return instance(request)

Caching Strategy

"""
production_cache.py - Production caching strategy.
"""

import redis
import pickle
import hashlib
from typing import Any, Optional

class ProductionCache:
    """Production caching with Redis."""

    def __init__(self, redis_url: str, ttl: int = 3600):
        self.redis_client = redis.from_url(redis_url)
        self.ttl = ttl

    def _generate_key(self, prefix: str, data: Any) -> str:
        """Generate cache key."""
        data_str = str(data)
        hash_obj = hashlib.md5(data_str.encode())
        return f"{prefix}:{hash_obj.hexdigest()}"

    def get(self, key: str) -> Optional[Any]:
        """Get from cache."""
        try:
            data = self.redis_client.get(key)
            if data:
                return pickle.loads(data)
        except Exception as e:
            logger.error(f"Cache get error: {e}")
        return None

    def set(self, key: str, value: Any, ttl: int = None):
        """Set in cache."""
        try:
            ttl = ttl or self.ttl
            data = pickle.dumps(value)
            self.redis_client.setex(key, ttl, data)
        except Exception as e:
            logger.error(f"Cache set error: {e}")

    def delete(self, key: str):
        """Delete from cache."""
        try:
            self.redis_client.delete(key)
        except Exception as e:
            logger.error(f"Cache delete error: {e}")

    def clear_all(self):
        """Clear all cache."""
        try:
            self.redis_client.flushdb()
            logger.info("Cleared all cache")
        except Exception as e:
            logger.error(f"Cache clear error: {e}")


# Initialize cache
cache = ProductionCache(
    redis_url=config.redis_url,
    ttl=config.cache_ttl
) if config.redis_url else None

Security Hardening

Authentication and Authorization

"""
production_security.py - Production security implementation.
"""

from fastapi import HTTPException, Security, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt
from datetime import datetime, timedelta
from typing import Dict, Optional

security = HTTPBearer()


class SecurityManager:
    """Manage authentication and authorization."""

    def __init__(self, jwt_secret: str):
        self.jwt_secret = jwt_secret
        self.algorithm = "HS256"

    def create_token(self, user_id: str, role: str) -> str:
        """Create JWT token."""
        payload = {
            "user_id": user_id,
            "role": role,
            "exp": datetime.utcnow() + timedelta(hours=24)
        }
        return jwt.encode(payload, self.jwt_secret, algorithm=self.algorithm)

    def verify_token(self, token: str) -> Dict:
        """Verify JWT token."""
        try:
            payload = jwt.decode(
                token,
                self.jwt_secret,
                algorithms=[self.algorithm]
            )
            return payload
        except jwt.ExpiredSignatureError:
            raise HTTPException(status_code=401, detail="Token expired")
        except jwt.InvalidTokenError:
            raise HTTPException(status_code=401, detail="Invalid token")

    def require_auth(
        self,
        credentials: HTTPAuthorizationCredentials = Security(security)
    ) -> Dict:
        """Require authentication."""
        token = credentials.credentials
        return self.verify_token(token)

    def require_role(self, required_role: str):
        """Require specific role."""
        def decorator(user: Dict = Depends(self.require_auth)):
            if user.get("role") != required_role:
                raise HTTPException(
                    status_code=403,
                    detail="Insufficient permissions"
                )
            return user
        return decorator


# Initialize security
security_manager = SecurityManager(jwt_secret=config.jwt_secret)


# Use in endpoints
@app.post("/execute")
async def execute_task(
    request: Dict,
    user: Dict = Depends(security_manager.require_auth)
):
    """Execute task with authentication."""
    # Process request
    pass

Rate Limiting

"""
rate_limiting.py - Production rate limiting.
"""

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from fastapi import Request

# Initialize rate limiter
limiter = Limiter(key_func=get_remote_address)

# Add to FastAPI app
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)


@app.post("/execute")
@limiter.limit("60/minute")  # 60 requests per minute
async def execute_task(request: Request):
    """Execute task with rate limiting."""
    # Process request
    pass

High Availability

Multi-Region Deployment

"""
high_availability.py - High availability implementation.
"""

from typing import List, Dict, Any
import logging

logger = logging.getLogger(__name__)


class MultiRegionDeployment:
    """Manage multi-region deployment for high availability."""

    def __init__(self, regions: List[str]):
        self.regions = regions
        self.region_instances: Dict[str, Any] = {}
        self.health_status: Dict[str, bool] = {}

    def deploy_to_region(self, region: str, team):
        """Deploy team to specific region."""
        logger.info(f"Deploying to region: {region}")
        self.region_instances[region] = team
        self.health_status[region] = True

    def get_healthy_regions(self) -> List[str]:
        """Get list of healthy regions."""
        return [
            region
            for region, healthy in self.health_status.items()
            if healthy
        ]

    def route_to_nearest_region(self, user_location: str):
        """Route request to nearest healthy region."""
        # Simplified region routing logic
        region_distances = {
            "us-east-1": self._calculate_distance(user_location, "us-east-1"),
            "us-west-2": self._calculate_distance(user_location, "us-west-2"),
            "eu-west-1": self._calculate_distance(user_location, "eu-west-1")
        }

        # Sort by distance
        sorted_regions = sorted(
            region_distances.items(),
            key=lambda x: x[1]
        )

        # Find nearest healthy region
        for region, _ in sorted_regions:
            if region in self.health_status and self.health_status[region]:
                logger.info(f"Routing to region: {region}")
                return self.region_instances[region]

        raise RuntimeError("No healthy regions available")

    def _calculate_distance(self, from_loc: str, to_region: str) -> float:
        """Calculate distance between locations."""
        # Simplified distance calculation
        # In production, use actual geolocation
        return 0.0

    def perform_health_check(self, region: str) -> bool:
        """Perform health check on region."""
        try:
            team = self.region_instances[region]
            # Test basic functionality
            result = team({
                "messages": [HumanMessage(content="health check")]
            })
            self.health_status[region] = True
            return True
        except Exception as e:
            logger.error(f"Region {region} health check failed: {e}")
            self.health_status[region] = False
            return False

    def failover_to_backup(self, failed_region: str):
        """Failover to backup region."""
        logger.warning(f"Initiating failover from {failed_region}")

        # Mark failed region as unhealthy
        self.health_status[failed_region] = False

        # Find healthy backup region
        healthy_regions = self.get_healthy_regions()

        if not healthy_regions:
            raise RuntimeError("No healthy backup regions available")

        backup_region = healthy_regions[0]
        logger.info(f"Failed over to backup region: {backup_region}")

        return backup_region

Database Replication

# database_replication.yaml - PostgreSQL replication setup
apiVersion: v1
kind: ConfigMap
metadata:
  name: postgres-replication-config
  namespace: production
data:
  postgresql.conf: |
    # Replication settings
    wal_level = replica
    max_wal_senders = 10
    max_replication_slots = 10
    hot_standby = on

  pg_hba.conf: |
    # Replication connections
    host replication replicator 0.0.0.0/0 md5

---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: postgres-primary
  namespace: production
spec:
  serviceName: postgres-primary
  replicas: 1
  selector:
    matchLabels:
      app: postgres
      role: primary
  template:
    metadata:
      labels:
        app: postgres
        role: primary
    spec:
      containers:
      - name: postgres
        image: postgres:15
        env:
        - name: POSTGRES_USER
          value: mcp_user
        - name: POSTGRES_PASSWORD
          valueFrom:
            secretKeyRef:
              name: postgres-secrets
              key: password
        - name: POSTGRES_DB
          value: mcp_db
        ports:
        - containerPort: 5432
          name: postgres
        volumeMounts:
        - name: data
          mountPath: /var/lib/postgresql/data
        - name: config
          mountPath: /etc/postgresql
      volumes:
      - name: config
        configMap:
          name: postgres-replication-config
  volumeClaimTemplates:
  - metadata:
      name: data
    spec:
      accessModes: ["ReadWriteOnce"]
      resources:
        requests:
          storage: 100Gi

---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: postgres-replica
  namespace: production
spec:
  serviceName: postgres-replica
  replicas: 2
  selector:
    matchLabels:
      app: postgres
      role: replica
  template:
    metadata:
      labels:
        app: postgres
        role: replica
    spec:
      containers:
      - name: postgres
        image: postgres:15
        env:
        - name: POSTGRES_USER
          value: mcp_user
        - name: POSTGRES_PASSWORD
          valueFrom:
            secretKeyRef:
              name: postgres-secrets
              key: password
        - name: POSTGRES_PRIMARY_HOST
          value: postgres-primary
        ports:
        - containerPort: 5432
          name: postgres
        volumeMounts:
        - name: data
          mountPath: /var/lib/postgresql/data
  volumeClaimTemplates:
  - metadata:
      name: data
    spec:
      accessModes: ["ReadWriteOnce"]
      resources:
        requests:
          storage: 100Gi

Load Balancer Configuration

"""
advanced_load_balancer.py - Advanced load balancing strategies.
"""

import time
from typing import List, Dict, Any
from collections import defaultdict
import heapq


class WeightedLoadBalancer:
    """Weighted round-robin load balancer."""

    def __init__(self, instances: List[tuple]):
        """
        Initialize with weighted instances.

        Args:
            instances: List of (instance, weight) tuples
        """
        self.instances = instances
        self.current_weight = 0
        self.current_index = -1
        self.max_weight = max(weight for _, weight in instances)
        self.gcd_weight = self._gcd_weights()

    def _gcd_weights(self) -> int:
        """Calculate GCD of all weights."""
        from math import gcd
        weights = [weight for _, weight in self.instances]
        result = weights[0]
        for weight in weights[1:]:
            result = gcd(result, weight)
        return result

    def get_instance(self):
        """Get next instance using weighted round-robin."""
        while True:
            self.current_index = (self.current_index + 1) % len(self.instances)

            if self.current_index == 0:
                self.current_weight = self.current_weight - self.gcd_weight
                if self.current_weight <= 0:
                    self.current_weight = self.max_weight

            instance, weight = self.instances[self.current_index]
            if weight >= self.current_weight:
                return instance


class LeastConnectionsLoadBalancer:
    """Least connections load balancer."""

    def __init__(self, instances: List):
        self.instances = instances
        self.connections: Dict[int, int] = defaultdict(int)

    def get_instance(self):
        """Get instance with least connections."""
        min_connections = min(
            self.connections[id(instance)]
            for instance in self.instances
        )

        for instance in self.instances:
            if self.connections[id(instance)] == min_connections:
                self.connections[id(instance)] += 1
                return instance

    def release_instance(self, instance):
        """Release instance connection."""
        instance_id = id(instance)
        if instance_id in self.connections:
            self.connections[instance_id] = max(
                0,
                self.connections[instance_id] - 1
            )


class ConsistentHashLoadBalancer:
    """Consistent hashing load balancer."""

    def __init__(self, instances: List, virtual_nodes: int = 150):
        self.virtual_nodes = virtual_nodes
        self.ring: Dict[int, Any] = {}
        self.sorted_keys: List[int] = []

        for instance in instances:
            self.add_instance(instance)

    def _hash(self, key: str) -> int:
        """Hash function."""
        import hashlib
        return int(hashlib.md5(key.encode()).hexdigest(), 16)

    def add_instance(self, instance):
        """Add instance to hash ring."""
        for i in range(self.virtual_nodes):
            virtual_key = f"{id(instance)}:{i}"
            hash_val = self._hash(virtual_key)
            self.ring[hash_val] = instance
            heapq.heappush(self.sorted_keys, hash_val)

    def get_instance(self, key: str):
        """Get instance for given key."""
        if not self.ring:
            return None

        hash_val = self._hash(key)

        # Find first node >= hash_val
        for node_hash in self.sorted_keys:
            if node_hash >= hash_val:
                return self.ring[node_hash]

        # Wrap around to first node
        return self.ring[self.sorted_keys[0]]

Disaster Recovery

Backup Strategy

"""
backup_strategy.py - Comprehensive backup implementation.
"""

import os
import shutil
import boto3
from datetime import datetime, timedelta
from typing import List, Dict, Any
import logging

logger = logging.getLogger(__name__)


class BackupManager:
    """Manage backups for disaster recovery."""

    def __init__(
        self,
        backup_dir: str = "/backups",
        s3_bucket: str = None,
        retention_days: int = 30
    ):
        self.backup_dir = backup_dir
        self.s3_bucket = s3_bucket
        self.retention_days = retention_days

        if s3_bucket:
            self.s3_client = boto3.client('s3')

        os.makedirs(backup_dir, exist_ok=True)

    def backup_database(self, db_url: str) -> str:
        """Backup database."""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        backup_file = f"{self.backup_dir}/db_backup_{timestamp}.sql"

        logger.info(f"Starting database backup: {backup_file}")

        try:
            # Use pg_dump for PostgreSQL
            import subprocess
            subprocess.run([
                "pg_dump",
                db_url,
                "-f", backup_file
            ], check=True)

            logger.info(f"Database backup completed: {backup_file}")

            # Upload to S3 if configured
            if self.s3_bucket:
                self._upload_to_s3(backup_file)

            return backup_file

        except Exception as e:
            logger.error(f"Database backup failed: {e}")
            raise

    def backup_rl_data(self, rl_data_dir: str) -> str:
        """Backup RL Q-table and training data."""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        backup_file = f"{self.backup_dir}/rl_backup_{timestamp}.tar.gz"

        logger.info(f"Starting RL data backup: {backup_file}")

        try:
            import tarfile
            with tarfile.open(backup_file, "w:gz") as tar:
                tar.add(rl_data_dir, arcname="rl_data")

            logger.info(f"RL data backup completed: {backup_file}")

            # Upload to S3 if configured
            if self.s3_bucket:
                self._upload_to_s3(backup_file)

            return backup_file

        except Exception as e:
            logger.error(f"RL data backup failed: {e}")
            raise

    def backup_configuration(self, config_dir: str) -> str:
        """Backup configuration files."""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        backup_file = f"{self.backup_dir}/config_backup_{timestamp}.tar.gz"

        logger.info(f"Starting configuration backup: {backup_file}")

        try:
            import tarfile
            with tarfile.open(backup_file, "w:gz") as tar:
                tar.add(config_dir, arcname="config")

            logger.info(f"Configuration backup completed: {backup_file}")

            # Upload to S3 if configured
            if self.s3_bucket:
                self._upload_to_s3(backup_file)

            return backup_file

        except Exception as e:
            logger.error(f"Configuration backup failed: {e}")
            raise

    def _upload_to_s3(self, file_path: str):
        """Upload backup to S3."""
        try:
            file_name = os.path.basename(file_path)
            s3_key = f"backups/{file_name}"

            logger.info(f"Uploading to S3: {s3_key}")

            self.s3_client.upload_file(
                file_path,
                self.s3_bucket,
                s3_key
            )

            logger.info(f"S3 upload completed: {s3_key}")

        except Exception as e:
            logger.error(f"S3 upload failed: {e}")
            raise

    def list_backups(self) -> List[Dict[str, Any]]:
        """List all backups."""
        backups = []

        for file_name in os.listdir(self.backup_dir):
            file_path = os.path.join(self.backup_dir, file_name)
            if os.path.isfile(file_path):
                stat = os.stat(file_path)
                backups.append({
                    "file": file_name,
                    "path": file_path,
                    "size": stat.st_size,
                    "created": datetime.fromtimestamp(stat.st_ctime)
                })

        return sorted(backups, key=lambda x: x["created"], reverse=True)

    def cleanup_old_backups(self):
        """Remove backups older than retention period."""
        cutoff_date = datetime.now() - timedelta(days=self.retention_days)

        logger.info(f"Cleaning up backups older than {cutoff_date}")

        for backup in self.list_backups():
            if backup["created"] < cutoff_date:
                logger.info(f"Removing old backup: {backup['file']}")
                os.remove(backup["path"])

    def restore_database(self, backup_file: str, db_url: str):
        """Restore database from backup."""
        logger.info(f"Restoring database from: {backup_file}")

        try:
            import subprocess
            subprocess.run([
                "psql",
                db_url,
                "-f", backup_file
            ], check=True)

            logger.info("Database restore completed")

        except Exception as e:
            logger.error(f"Database restore failed: {e}")
            raise

    def restore_rl_data(self, backup_file: str, target_dir: str):
        """Restore RL data from backup."""
        logger.info(f"Restoring RL data from: {backup_file}")

        try:
            import tarfile
            with tarfile.open(backup_file, "r:gz") as tar:
                tar.extractall(path=target_dir)

            logger.info("RL data restore completed")

        except Exception as e:
            logger.error(f"RL data restore failed: {e}")
            raise


# Automated backup schedule
def schedule_backups():
    """Schedule automated backups."""
    import schedule

    backup_manager = BackupManager(
        backup_dir="/backups",
        s3_bucket="my-mcp-backups",
        retention_days=30
    )

    # Daily database backup at 2 AM
    schedule.every().day.at("02:00").do(
        backup_manager.backup_database,
        db_url=config.database_url
    )

    # Daily RL data backup at 3 AM
    schedule.every().day.at("03:00").do(
        backup_manager.backup_rl_data,
        rl_data_dir="/app/rl_data"
    )

    # Weekly configuration backup on Sunday at 1 AM
    schedule.every().sunday.at("01:00").do(
        backup_manager.backup_configuration,
        config_dir="/app/config"
    )

    # Daily cleanup of old backups at 4 AM
    schedule.every().day.at("04:00").do(
        backup_manager.cleanup_old_backups
    )

    logger.info("Backup schedule configured")

    # Run scheduled tasks
    while True:
        schedule.run_pending()
        time.sleep(60)