⚙️ Session 3 Advanced: Production Deployment Strategies¶

# Your main configuration orchestrator - Complete implementation
class Config:
    """Main configuration class for LangChain MCP integration."""

    # API Keys from environment variables - Security first
    OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
    ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")

    # Environment detection - Deployment context awareness
    ENVIRONMENT = os.getenv("ENVIRONMENT", "development")
    DEBUG = os.getenv("DEBUG", "false").lower() == "true"

    # LLM Configuration with environment overrides - Flexibility with defaults
    LLM = LLMConfig(
        provider=os.getenv("LLM_PROVIDER", "openai"),
        model=os.getenv("LLM_MODEL", "gpt-4"),
        temperature=float(os.getenv("LLM_TEMPERATURE", "0.7")),
        max_tokens=int(os.getenv("LLM_MAX_TOKENS", "2000")),
        timeout=int(os.getenv("LLM_TIMEOUT", "60"))
    )

    # MCP Server Registry - Your digital workforce roster
    MCP_SERVERS = [
        MCPServerConfig(
            name="weather",
            command="python",
            args=["mcp_servers/weather_server.py"],
            description="Weather information and forecasts",
            timeout=int(os.getenv("WEATHER_SERVER_TIMEOUT", "30")),
            retry_attempts=int(os.getenv("WEATHER_SERVER_RETRIES", "3"))
        ),
        MCPServerConfig(
            name="filesystem",
            command="python",
            args=["mcp_servers/filesystem_server.py"],
            description="Secure file system operations",
            timeout=int(os.getenv("FS_SERVER_TIMEOUT", "45")),
            retry_attempts=int(os.getenv("FS_SERVER_RETRIES", "2"))
        ),
        MCPServerConfig(
            name="database",
            command="python",
            args=["mcp_servers/database_server.py"],
            description="Database query and manipulation",
            timeout=int(os.getenv("DB_SERVER_TIMEOUT", "60")),
            retry_attempts=int(os.getenv("DB_SERVER_RETRIES", "3"))
        )
    ]

    # Agent Configuration - Intelligence parameters with production settings
    AGENT_CONFIG = {
        "max_iterations": int(os.getenv("MAX_ITERATIONS", "10")),
        "verbose": os.getenv("VERBOSE", "true").lower() == "true",
        "temperature": float(os.getenv("AGENT_TEMPERATURE", "0.7")),
        "timeout": int(os.getenv("AGENT_TIMEOUT", "300")),  # 5 minutes
        "retry_attempts": int(os.getenv("AGENT_RETRY_ATTEMPTS", "2")),
        "parallel_execution": os.getenv("PARALLEL_EXECUTION", "false").lower() == "true"
    }

    # Production Monitoring Configuration
    MONITORING = {
        "enable_metrics": os.getenv("ENABLE_METRICS", "true").lower() == "true",
        "metrics_port": int(os.getenv("METRICS_PORT", "8080")),
        "health_check_interval": int(os.getenv("HEALTH_CHECK_INTERVAL", "30")),
        "log_level": os.getenv("LOG_LEVEL", "INFO"),
        "structured_logging": os.getenv("STRUCTURED_LOGGING", "true").lower() == "true"
    }

    # Performance Configuration
    PERFORMANCE = {
        "connection_pool_size": int(os.getenv("CONNECTION_POOL_SIZE", "10")),
        "request_timeout": int(os.getenv("REQUEST_TIMEOUT", "30")),
        "circuit_breaker_enabled": os.getenv("CIRCUIT_BREAKER", "true").lower() == "true",
        "rate_limiting_enabled": os.getenv("RATE_LIMITING", "false").lower() == "true",
        "cache_enabled": os.getenv("CACHE_ENABLED", "true").lower() == "true"
    }

# utils/monitoring.py - Production observability system
import logging
import time
import json
from typing import Dict, Any, Optional
from dataclasses import dataclass, asdict
from datetime import datetime
import asyncio
from prometheus_client import Counter, Histogram, Gauge, start_http_server

# Production metrics collection
AGENT_REQUESTS = Counter('agent_requests_total', 'Total agent requests', ['status', 'agent_type'])
AGENT_DURATION = Histogram('agent_request_duration_seconds', 'Agent request duration')
MCP_TOOL_CALLS = Counter('mcp_tool_calls_total', 'Total MCP tool calls', ['server', 'tool', 'status'])
ACTIVE_AGENTS = Gauge('active_agents_current', 'Currently active agents')
SERVER_HEALTH = Gauge('mcp_server_health', 'MCP server health status', ['server'])

@dataclass
class RequestMetrics:
    """Structured metrics for agent requests."""
    request_id: str
    agent_type: str
    query: str
    start_time: float
    end_time: Optional[float] = None
    success: bool = False
    error: Optional[str] = None
    tools_used: list = None
    execution_time: Optional[float] = None

class ProductionMonitor:
    """Enterprise monitoring and observability for LangChain-MCP systems."""

    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.logger = self._setup_structured_logging()
        self.active_requests: Dict[str, RequestMetrics] = {}

        if config.get("enable_metrics", True):
            self._start_metrics_server()

    def _setup_structured_logging(self) -> logging.Logger:
        """Configure structured logging for production observability."""
        logger = logging.getLogger("langchain_mcp_production")
        logger.setLevel(getattr(logging, self.config.get("log_level", "INFO")))

        # Structured logging formatter
        if self.config.get("structured_logging", True):
            formatter = logging.Formatter(
                fmt='{"timestamp":"%(asctime)s","level":"%(levelname)s",'
                    '"component":"%(name)s","message":"%(message)s",'
                    '"environment":"' + os.getenv("ENVIRONMENT", "unknown") + '"}'
            )
        else:
            formatter = logging.Formatter(
                fmt='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
            )

        handler = logging.StreamHandler()
        handler.setFormatter(formatter)
        logger.addHandler(handler)

        return logger

    def _start_metrics_server(self):
        """Start Prometheus metrics server for monitoring integration."""
        port = self.config.get("metrics_port", 8080)
        start_http_server(port)
        self.logger.info(f"Metrics server started on port {port}")

# utils/resilience.py - Enterprise resilience patterns
import asyncio
import functools
from typing import Callable, Any, Optional, Dict
from dataclasses import dataclass
from enum import Enum

class CircuitBreakerState(Enum):
    """Circuit breaker states for fault tolerance."""
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Failure mode - requests fail fast
    HALF_OPEN = "half_open"  # Testing recovery

@dataclass
class CircuitBreakerConfig:
    """Configuration for circuit breaker behavior."""
    failure_threshold: int = 5
    recovery_timeout: int = 60
    success_threshold: int = 2
    timeout: int = 30

class CircuitBreaker:
    """Production circuit breaker for MCP server protection."""

    def __init__(self, config: CircuitBreakerConfig, name: str = "default"):
        self.config = config
        self.name = name
        self.state = CircuitBreakerState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time = 0

    async def call(self, func: Callable, *args, **kwargs) -> Any:
        """Execute function with circuit breaker protection."""
        if self.state == CircuitBreakerState.OPEN:
            if time.time() - self.last_failure_time > self.config.recovery_timeout:
                self.state = CircuitBreakerState.HALF_OPEN
                self.success_count = 0
            else:
                raise CircuitBreakerOpenError(f"Circuit breaker {self.name} is open")

        try:
            result = await asyncio.wait_for(func(*args, **kwargs), timeout=self.config.timeout)
            await self._record_success()
            return result

        except Exception as e:
            await self._record_failure()
            raise e

    async def _record_success(self):
        """Record successful execution and potentially close circuit."""
        if self.state == CircuitBreakerState.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.config.success_threshold:
                self.state = CircuitBreakerState.CLOSED
                self.failure_count = 0
        elif self.state == CircuitBreakerState.CLOSED:
            self.failure_count = 0

    async def _record_failure(self):
        """Record failure and potentially open circuit."""
        self.failure_count += 1
        self.last_failure_time = time.time()

        if self.failure_count >= self.config.failure_threshold:
            self.state = CircuitBreakerState.OPEN

class ProductionAgentFactory:
    """Factory for creating production-ready LangChain-MCP agents."""

    def __init__(self, config: Config, monitor: ProductionMonitor):
        self.config = config
        self.monitor = monitor
        self.circuit_breakers: Dict[str, CircuitBreaker] = {}
        self.agent_pool: Dict[str, Any] = {}

    async def create_resilient_agent(self, agent_type: str = "multi-tool") -> 'ResilientAgent':
        """Create production agent with full resilience features."""

        # Initialize MCP server manager with circuit breakers
        mcp_manager = await self._create_resilient_mcp_manager()

        # Create agent based on type
        if agent_type == "multi-tool":
            base_agent = MultiToolMCPAgent(mcp_manager)
        elif agent_type == "workflow":
            base_agent = ResearchWorkflow(mcp_manager)
        else:
            raise ValueError(f"Unknown agent type: {agent_type}")

        # Wrap in resilience layer
        resilient_agent = ResilientAgent(
            base_agent=base_agent,
            monitor=self.monitor,
            circuit_breakers=self.circuit_breakers,
            config=self.config
        )

        await resilient_agent.initialize()
        return resilient_agent

    async def _create_resilient_mcp_manager(self) -> 'ResilientMCPManager':
        """Create MCP manager with production resilience features."""

        # Create circuit breakers for each server
        for server_config in self.config.MCP_SERVERS:
            breaker_config = CircuitBreakerConfig(
                failure_threshold=server_config.retry_attempts + 2,
                recovery_timeout=server_config.timeout * 2,
                timeout=server_config.timeout
            )
            self.circuit_breakers[server_config.name] = CircuitBreaker(
                breaker_config,
                name=f"mcp_{server_config.name}"
            )

        return ResilientMCPManager(
            server_configs=self.config.MCP_SERVERS,
            circuit_breakers=self.circuit_breakers,
            monitor=self.monitor
        )

class ResilientAgent:
    """Production-ready agent with comprehensive error handling and monitoring."""

    def __init__(self, base_agent, monitor: ProductionMonitor,
                 circuit_breakers: Dict[str, CircuitBreaker], config: Config):
        self.base_agent = base_agent
        self.monitor = monitor
        self.circuit_breakers = circuit_breakers
        self.config = config
        self.request_queue = asyncio.Queue()

    async def process_request(self, query: str, context: Dict[str, Any] = None) -> Dict[str, Any]:
        """Process request with full production monitoring and resilience."""
        request_id = f"req_{int(time.time() * 1000)}"

        # Start request tracking
        metrics = RequestMetrics(
            request_id=request_id,
            agent_type=type(self.base_agent).__name__,
            query=query,
            start_time=time.time()
        )

        self.monitor.active_requests[request_id] = metrics
        ACTIVE_AGENTS.inc()

        try:
            # Execute with timeout and monitoring
            result = await asyncio.wait_for(
                self._execute_with_retries(query, context),
                timeout=self.config.AGENT_CONFIG.get("timeout", 300)
            )

            # Record success metrics
            metrics.success = True
            metrics.end_time = time.time()
            metrics.execution_time = metrics.end_time - metrics.start_time
            metrics.tools_used = result.get("tools_used", [])

            AGENT_REQUESTS.labels(status="success", agent_type=metrics.agent_type).inc()
            AGENT_DURATION.observe(metrics.execution_time)

            self.monitor.logger.info(
                f"Request completed successfully",
                extra={"request_metrics": asdict(metrics)}
            )

            return result

        except Exception as e:
            # Record failure metrics
            metrics.success = False
            metrics.end_time = time.time()
            metrics.execution_time = metrics.end_time - metrics.start_time
            metrics.error = str(e)

            AGENT_REQUESTS.labels(status="error", agent_type=metrics.agent_type).inc()

            self.monitor.logger.error(
                f"Request failed: {str(e)}",
                extra={"request_metrics": asdict(metrics)}
            )

            return {
                "success": False,
                "request_id": request_id,
                "error": str(e),
                "execution_time": metrics.execution_time
            }

        finally:
            # Cleanup tracking
            del self.monitor.active_requests[request_id]
            ACTIVE_AGENTS.dec()

    async def _execute_with_retries(self, query: str, context: Dict[str, Any] = None) -> Dict[str, Any]:
        """Execute agent with retry logic and circuit breaker protection."""
        max_retries = self.config.AGENT_CONFIG.get("retry_attempts", 2)

        for attempt in range(max_retries + 1):
            try:
                if hasattr(self.base_agent, 'run'):
                    result = await self.base_agent.run(query, context)
                elif hasattr(self.base_agent, 'run_research'):
                    result = await self.base_agent.run_research(query)
                else:
                    raise ValueError("Agent must implement 'run' or 'run_research' method")

                return result

            except Exception as e:
                if attempt < max_retries:
                    wait_time = 2 ** attempt  # Exponential backoff
                    self.monitor.logger.warning(
                        f"Agent execution failed (attempt {attempt + 1}), retrying in {wait_time}s: {str(e)}"
                    )
                    await asyncio.sleep(wait_time)
                else:
                    raise e

# deployment/docker-compose.yml - Production deployment template
"""
version: '3.8'

services:
  langchain-mcp-agent:
    build: .
    ports:
      - "8000:8000"
      - "8080:8080"  # Metrics port
    environment:
      - ENVIRONMENT=production
      - LOG_LEVEL=INFO
      - STRUCTURED_LOGGING=true
      - ENABLE_METRICS=true
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    volumes:
      - ./config:/app/config
      - ./logs:/app/logs
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3
    deploy:
      resources:
        limits:
          cpus: '2.0'
          memory: 4G
        reservations:
          cpus: '0.5'
          memory: 1G

  prometheus:
    image: prom/prometheus:latest
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
    command:
      - '--config.file=/etc/prometheus/prometheus.yml'
      - '--storage.tsdb.path=/prometheus'
      - '--web.console.libraries=/etc/prometheus/console_libraries'
      - '--web.console.templates=/etc/prometheus/consoles'

  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin
    volumes:
      - grafana-storage:/var/lib/grafana
      - ./grafana/dashboards:/etc/grafana/provisioning/dashboards
      - ./grafana/datasources:/etc/grafana/provisioning/datasources

volumes:
  grafana-storage:
"""

# deployment/kubernetes.yml - Kubernetes deployment template
"""
apiVersion: apps/v1
kind: Deployment
metadata:
  name: langchain-mcp-agent
  labels:
    app: langchain-mcp-agent
spec:
  replicas: 3
  selector:
    matchLabels:
      app: langchain-mcp-agent
  template:
    metadata:
      labels:
        app: langchain-mcp-agent
    spec:
      containers:
      - name: agent
        image: langchain-mcp-agent:latest
        ports:
        - containerPort: 8000
        - containerPort: 8080
        env:
        - name: ENVIRONMENT
          value: "production"
        - name: OPENAI_API_KEY
          valueFrom:
            secretKeyRef:
              name: api-secrets
              key: openai-key
        resources:
          limits:
            cpu: 2000m
            memory: 4Gi
          requests:
            cpu: 500m
            memory: 1Gi
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 5
"""

# utils/performance.py - Production performance optimizations
import asyncio
import aioredis
from typing import Optional, Any, Dict
import hashlib
import json
import time

class PerformanceOptimizer:
    """Production performance optimization for LangChain-MCP systems."""

    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.cache_client: Optional[aioredis.Redis] = None
        self.connection_pool = None

    async def initialize(self):
        """Initialize performance optimization components."""
        if self.config.get("cache_enabled", True):
            await self._initialize_cache()

        if self.config.get("connection_pool_size", 10) > 0:
            await self._initialize_connection_pool()

    async def _initialize_cache(self):
        """Initialize Redis cache for response caching."""
        redis_url = os.getenv("REDIS_URL", "redis://localhost:6379")
        self.cache_client = await aioredis.from_url(redis_url)

    async def cached_execution(self, cache_key: str, func: Callable, ttl: int = 300) -> Any:
        """Execute function with caching support."""
        if not self.cache_client:
            return await func()

        # Try cache first
        cached_result = await self.cache_client.get(cache_key)
        if cached_result:
            return json.loads(cached_result)

        # Execute and cache
        result = await func()
        await self.cache_client.setex(
            cache_key,
            ttl,
            json.dumps(result, default=str)
        )
        return result

    def generate_cache_key(self, query: str, agent_type: str, context: Dict = None) -> str:
        """Generate cache key for query results."""
        key_data = {
            "query": query,
            "agent_type": agent_type,
            "context": context or {}
        }
        key_string = json.dumps(key_data, sort_keys=True)
        return f"agent_response:{hashlib.md5(key_string.encode()).hexdigest()}"

# utils/health.py - Production health monitoring
class HealthChecker:
    """Comprehensive health checking for production deployments."""

    def __init__(self, mcp_manager, monitor: ProductionMonitor):
        self.mcp_manager = mcp_manager
        self.monitor = monitor

    async def comprehensive_health_check(self) -> Dict[str, Any]:
        """Perform comprehensive system health check."""
        health_status = {
            "timestamp": datetime.utcnow().isoformat(),
            "overall_status": "healthy",
            "components": {}
        }

        # Check MCP servers
        mcp_health = await self._check_mcp_servers()
        health_status["components"]["mcp_servers"] = mcp_health

        # Check LLM connectivity
        llm_health = await self._check_llm_connectivity()
        health_status["components"]["llm"] = llm_health

        # Check system resources
        resource_health = await self._check_system_resources()
        health_status["components"]["resources"] = resource_health

        # Determine overall status
        component_statuses = [comp["status"] for comp in health_status["components"].values()]
        if "unhealthy" in component_statuses:
            health_status["overall_status"] = "unhealthy"
        elif "degraded" in component_statuses:
            health_status["overall_status"] = "degraded"

        return health_status

    async def _check_mcp_servers(self) -> Dict[str, Any]:
        """Check health of all MCP servers."""
        servers = {}
        healthy_count = 0

        for server_name in self.mcp_manager.server_configs.keys():
            try:
                adapter = await self.mcp_manager.get_adapter(server_name)
                if adapter and self.mcp_manager.health_status.get(server_name, False):
                    servers[server_name] = {"status": "healthy", "response_time": None}
                    healthy_count += 1
                else:
                    servers[server_name] = {"status": "unhealthy", "error": "Server unavailable"}
            except Exception as e:
                servers[server_name] = {"status": "unhealthy", "error": str(e)}

        total_servers = len(self.mcp_manager.server_configs)
        overall_status = "healthy" if healthy_count == total_servers else \
                        "degraded" if healthy_count > 0 else "unhealthy"

        return {
            "status": overall_status,
            "healthy_servers": healthy_count,
            "total_servers": total_servers,
            "servers": servers
        }