Configuration - YFinance MCP Server

Overview

The Financial MCP Server offers extensive configuration options to adapt to different use cases, from development to enterprise production environments.

Environment Variables

Core Configuration

Advanced Configuration

Configuration Examples

Development Environment

# .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

# 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

Docker Configuration

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

Server Customization

Logging Configuration

# 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())

CORS Configuration

# 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"]
)

Rate Limiting

# 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)

Caching Configuration

# 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

Tool Configuration

Enabling/Disabling Tools

# 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()

Custom Tool Parameters

# 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)

Security Configuration

API Key Management

# 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...

Access Control

# 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

Performance Tuning

Connection Pooling

# 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()

Memory Management

# 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

Monitoring Configuration

Health Checks

@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

Deployment-Specific Configuration

Railway Configuration

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

Docker Configuration

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"]

Next Steps

Deployment Guide

Deploy your configured server to production

Troubleshooting

Solve common configuration issues

API Reference

Explore all available endpoints

Architecture

Understand the server architecture

​Overview

​Environment Variables

​Core Configuration

​Advanced Configuration

​Configuration Examples

​Development Environment

​Production Environment

​Docker Configuration

​Server Customization

​Logging Configuration

​CORS Configuration

​Rate Limiting

​Caching Configuration

​Tool Configuration

​Enabling/Disabling Tools

​Custom Tool Parameters

​Security Configuration

​API Key Management

​Access Control

​Performance Tuning

​Connection Pooling

​Memory Management

​Monitoring Configuration

​Health Checks

​Deployment-Specific Configuration

​Railway Configuration

​Docker Configuration

​Next Steps

Deployment Guide

Troubleshooting

API Reference

Architecture

Overview

Environment Variables

Core Configuration

Advanced Configuration

Configuration Examples

Development Environment

Production Environment

Docker Configuration

Server Customization

Logging Configuration

CORS Configuration

Rate Limiting

Caching Configuration

Tool Configuration

Enabling/Disabling Tools

Custom Tool Parameters

Security Configuration

API Key Management

Access Control

Performance Tuning

Connection Pooling

Memory Management

Monitoring Configuration

Health Checks

Deployment-Specific Configuration

Railway Configuration

Docker Configuration

Next Steps