Configuration

# .env file for development
FMP_API_KEY=your_development_api_key
PORT=8001
HOST=127.0.0.1
LOG_LEVEL=DEBUG
CORS_ORIGINS=http://localhost:3000,http://localhost:3001
REQUEST_TIMEOUT=30
CACHE_TTL=60

# Production environment variables
FMP_API_KEY=your_production_api_key
PORT=8001
HOST=0.0.0.0
LOG_LEVEL=INFO
CORS_ORIGINS=https://yourdomain.com,https://app.yourdomain.com
REQUEST_TIMEOUT=15
CACHE_TTL=300
MAX_CONNECTIONS=200

version: '3.8'
services:
  financial-mcp:
    build: .
    environment:
      - FMP_API_KEY=${FMP_API_KEY}
      - PORT=8001
      - HOST=0.0.0.0
      - LOG_LEVEL=INFO
      - CORS_ORIGINS=*
      - REQUEST_TIMEOUT=30
      - CACHE_TTL=300
    ports:
      - "8001:8001"
    restart: unless-stopped

# main.py - Custom logging setup
import logging
import sys
from datetime import datetime

# Create custom formatter
class CustomFormatter(logging.Formatter):
    def format(self, record):
        record.timestamp = datetime.now().isoformat()
        return super().format(record)

# Configure logging
log_level = os.getenv("LOG_LEVEL", "INFO").upper()
logging.basicConfig(
    level=getattr(logging, log_level),
    format='%(timestamp)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.StreamHandler(sys.stdout),
        logging.FileHandler('financial-mcp.log')
    ]
)

# Set custom formatter
for handler in logging.getLogger().handlers:
    handler.setFormatter(CustomFormatter())

# main.py - Advanced CORS setup
from fastapi.middleware.cors import CORSMiddleware

# Parse CORS origins from environment
cors_origins = os.getenv("CORS_ORIGINS", "*").split(",")
cors_origins = [origin.strip() for origin in cors_origins]

app.add_middleware(
    CORSMiddleware,
    allow_origins=cors_origins,
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["*"],
    expose_headers=["X-Request-ID", "X-Response-Time"]
)

# Add rate limiting middleware
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.get("/api/v1/stock/{symbol}")
@limiter.limit("60/minute")  # 60 requests per minute
async def get_stock_quote(request: Request, symbol: str):
    return await stock_quote_handler(symbol)

# In-memory caching with TTL
from functools import lru_cache
import time
import asyncio

class TTLCache:
    def __init__(self, ttl_seconds=300):
        self.cache = {}
        self.ttl = ttl_seconds
    
    def get(self, key):
        if key in self.cache:
            value, timestamp = self.cache[key]
            if time.time() - timestamp < self.ttl:
                return value
            else:
                del self.cache[key]
        return None
    
    def set(self, key, value):
        self.cache[key] = (value, time.time())

# Global cache instance
cache = TTLCache(ttl_seconds=int(os.getenv("CACHE_TTL", "300")))

async def cached_fmp_request(endpoint):
    cached_result = cache.get(endpoint)
    if cached_result:
        return cached_result
    
    result = await fmp_request(endpoint)
    cache.set(endpoint, result)
    return result

# Configure which tools to load
ENABLED_YFINANCE_TOOLS = os.getenv("YFINANCE_TOOLS", "all").split(",")
ENABLED_FMP_TOOLS = os.getenv("FMP_TOOLS", "all").split(",")

def register_tools():
    # YFinance tools
    if "all" in ENABLED_YFINANCE_TOOLS or "quote" in ENABLED_YFINANCE_TOOLS:
        mcp.tool("get_stock_quote")(get_stock_quote)
    
    if "all" in ENABLED_YFINANCE_TOOLS or "overview" in ENABLED_YFINANCE_TOOLS:
        mcp.tool("get_company_overview")(get_company_overview)
    
    # FMP tools
    if "all" in ENABLED_FMP_TOOLS or "gainers" in ENABLED_FMP_TOOLS:
        mcp.tool("fmp_get_market_gainers")(fmp_get_market_gainers)
    
    # Continue for other tools...

register_tools()

# Customize default parameters for tools
DEFAULT_STOCK_LIMIT = int(os.getenv("DEFAULT_STOCK_LIMIT", "20"))
DEFAULT_CRYPTO_LIMIT = int(os.getenv("DEFAULT_CRYPTO_LIMIT", "50"))
DEFAULT_HISTORICAL_PERIOD = os.getenv("DEFAULT_HISTORICAL_PERIOD", "1y")

@mcp.tool("fmp_get_market_gainers")
async def fmp_get_market_gainers(limit: int = DEFAULT_STOCK_LIMIT):
    return await fmp_market_gainers(limit)

@mcp.tool("fmp_get_crypto_prices")
async def fmp_get_crypto_prices(limit: int = DEFAULT_CRYPTO_LIMIT):
    return await fmp_crypto_prices(limit)

# Support multiple FMP API keys for load balancing
FMP_API_KEYS = os.getenv("FMP_API_KEYS", "").split(",")
FMP_API_KEYS = [key.strip() for key in FMP_API_KEYS if key.strip()]

import random

def get_api_key():
    if not FMP_API_KEYS:
        return os.getenv("FMP_API_KEY", "demo")
    return random.choice(FMP_API_KEYS)

async def fmp_request(endpoint):
    api_key = get_api_key()
    url = f"https://financialmodelingprep.com/api/v3/{endpoint}?apikey={api_key}"
    # Make request...

# Restrict access to specific IP addresses
ALLOWED_IPS = os.getenv("ALLOWED_IPS", "").split(",")
ALLOWED_IPS = [ip.strip() for ip in ALLOWED_IPS if ip.strip()]

@app.middleware("http")
async def ip_whitelist_middleware(request: Request, call_next):
    client_ip = request.client.host
    
    if ALLOWED_IPS and client_ip not in ALLOWED_IPS:
        return JSONResponse(
            status_code=403,
            content={"error": "Access denied", "ip": client_ip}
        )
    
    response = await call_next(request)
    return response

# Optimize HTTP client for performance
import aiohttp
import asyncio

class HTTPClientManager:
    def __init__(self):
        self.session = None
    
    async def get_session(self):
        if self.session is None:
            timeout = aiohttp.ClientTimeout(
                total=int(os.getenv("REQUEST_TIMEOUT", "30"))
            )
            
            connector = aiohttp.TCPConnector(
                limit=int(os.getenv("MAX_CONNECTIONS", "100")),
                limit_per_host=int(os.getenv("MAX_CONNECTIONS_PER_HOST", "30")),
                keepalive_timeout=30,
                enable_cleanup_closed=True
            )
            
            self.session = aiohttp.ClientSession(
                connector=connector,
                timeout=timeout
            )
        
        return self.session
    
    async def close(self):
        if self.session:
            await self.session.close()

http_manager = HTTPClientManager()

# Use in startup/shutdown events
@app.on_event("shutdown")
async def shutdown_event():
    await http_manager.close()

# Monitor and optimize memory usage
import gc
import psutil
import os

def get_memory_usage():
    process = psutil.Process(os.getpid())
    memory_mb = process.memory_info().rss / 1024 / 1024
    return memory_mb

@app.middleware("http")
async def memory_monitor_middleware(request: Request, call_next):
    memory_before = get_memory_usage()
    
    response = await call_next(request)
    
    memory_after = get_memory_usage()
    memory_diff = memory_after - memory_before
    
    # Log high memory usage requests
    if memory_diff > 50:  # 50MB threshold
        logger.warning(f"High memory usage: {memory_diff:.2f}MB for {request.url}")
    
    # Trigger garbage collection if memory usage is high
    if memory_after > 500:  # 500MB threshold
        gc.collect()
    
    return response

@app.get("/api/v1/health")
async def health_check():
    health_data = {
        "status": "healthy",
        "timestamp": datetime.now().isoformat(),
        "version": "1.0.0",
        "tools_available": 32,
        "data_sources": ["YFinance", "FMP"]
    }
    
    # Check FMP API connectivity
    try:
        test_response = await fmp_request("quote/AAPL")
        health_data["fmp_status"] = "connected"
    except Exception as e:
        health_data["fmp_status"] = "error"
        health_data["fmp_error"] = str(e)
    
    # Check YFinance connectivity
    try:
        import yfinance as yf
        ticker = yf.Ticker("AAPL")
        ticker.info
        health_data["yfinance_status"] = "connected"
    except Exception as e:
        health_data["yfinance_status"] = "error"
        health_data["yfinance_error"] = str(e)
    
    # System metrics
    health_data["memory_usage_mb"] = get_memory_usage()
    health_data["cache_size"] = len(cache.cache) if hasattr(cache, 'cache') else 0
    
    return health_data

{
  "build": {
    "builder": "NIXPACKS"
  },
  "deploy": {
    "restartPolicyType": "ON_FAILURE",
    "restartPolicyMaxRetries": 10,
    "healthcheckPath": "/api/v1/health",
    "healthcheckTimeout": 30
  }
}

FROM python:3.11-slim

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

# Create non-root user
RUN useradd --create-home --shell /bin/bash app

# Set working directory
WORKDIR /home/app

# Copy requirements and install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY --chown=app:app . .

# Switch to non-root user
USER app

# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
  CMD curl -f http://localhost:${PORT:-8001}/api/v1/health || exit 1

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