Skip to main content

​
System Overview

The Financial MCP Server is built as a hybrid platform supporting both Model Context Protocol (MCP) and REST API interfaces, providing seamless access to financial data from multiple sources.

​
Core Components

​
1. Protocol Handlers

​
MCP Protocol Handler

  • Purpose: Native AI agent integration
  • Technology: Model Context Protocol standard
  • Features: Structured tool calls, type safety, error handling
  • Use Case: Claude Desktop, custom AI agents

​
REST API Handler

  • Purpose: Traditional web/mobile application integration
  • Technology: FastAPI with automatic OpenAPI documentation
  • Features: HTTP endpoints, JSON responses, CORS support
  • Use Case: Frontend applications, mobile apps, webhooks

​
2. Application Core

​
Financial MCP Server (main.py)

# Hybrid server supporting both protocols
app = FastAPI(title="Financial MCP Server")
mcp = MCPServer("financial-data")

# Single server, dual protocols
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=PORT)
Responsibilities:
  • Protocol routing and handling
  • Server lifecycle management
  • Configuration management
  • Logging and monitoring

​
Tool Registry

# Centralized tool registration
@mcp.tool("get_stock_quote")
async def mcp_get_stock_quote(symbol: str):
    return await get_stock_quote(symbol)

@app.get("/api/v1/stock/{symbol}")
async def rest_get_stock_quote(symbol: str):
    return await get_stock_quote(symbol)
Features:
  • Single function, dual exposure (MCP + REST)
  • Consistent error handling
  • Parameter validation
  • Response formatting

​
3. Data Access Layer

​
YFinance Tools (tools/yfinance_tools.py)

# Native Python library integration
async def get_stock_quote(symbol: str) -> Dict[str, Any]:
    def _fetch_quote():
        ticker = yf.Ticker(symbol)
        return ticker.info
    
    return handle_yfinance_request(_fetch_quote)
Features:
  • Rate limiting protection
  • Automatic retry logic
  • Error handling and fallback
  • Data normalization

​
FMP Tools (tools/fmp_tools.py)

# Professional API integration
async def fmp_get_stock_quote(symbol: str) -> Dict[str, Any]:
    endpoint = f"quote/{symbol}"
    data = await fmp_request(endpoint)
    return format_response(data)
Features:
  • Professional API reliability
  • Structured data responses
  • Comprehensive error handling
  • Rate limit monitoring

​
Utility Layer (tools/fmp_utils.py)

# Shared utilities and helpers
async def fmp_request(endpoint: str, params: dict = None) -> dict:
    # Common request handling
    # Error processing
    # Response validation

​
Design Patterns

​
1. Adapter Pattern

Each data source implements a common interface: 
# Common interface for all data sources
class DataSource:
    async def get_quote(self, symbol: str) -> Dict[str, Any]
    async def get_company_info(self, symbol: str) -> Dict[str, Any]
    async def get_historical_data(self, symbol: str) -> Dict[str, Any]

​
2. Strategy Pattern

Different strategies for data retrieval: 
# Strategy selection based on requirements
def select_data_source(requirement: str) -> DataSource:
    if requirement == "free":
        return YFinanceAdapter()
    elif requirement == "professional":
        return FMPAdapter()
    elif requirement == "hybrid":
        return HybridAdapter()  # Try YFinance, fallback to FMP

​
3. Decorator Pattern

Cross-cutting concerns handled via decorators: 
@rate_limit_protection
@error_handling
@response_formatting
async def get_stock_quote(symbol: str):
    # Core business logic
    pass

​
Modular Structure

yfi-deployment/
├── main.py                 # Server entry point
├── api/
│   ├── __init__.py
│   └── rest_routes.py      # REST endpoint definitions
├── tools/
│   ├── __init__.py
│   ├── yfinance_tools.py   # YFinance integration (6 tools)
│   ├── fmp_tools.py        # FMP integration (26 tools)
│   └── fmp_utils.py        # Shared FMP utilities
├── requirements.txt        # Dependencies
├── Procfile               # Deployment configuration
└── docs/                  # Documentation (separate folder)

​
Benefits of Modular Design

  • Clear separation of concerns
  • Easy to locate and modify specific functionality
  • Reduced coupling between components
  • Add new data sources without affecting existing code
  • Independent scaling of different components
  • Easy to extend with additional tools
  • Unit test individual components in isolation
  • Mock external dependencies easily
  • Clear interfaces for testing
  • Tools can be used in both MCP and REST contexts
  • Utilities shared across different data sources
  • Easy integration into other projects

​
Data Flow

​
Request Processing Flow

​
Error Handling Flow

​
Performance Considerations

​
1. Asynchronous Processing

# Non-blocking I/O for concurrent requests
async def handle_multiple_requests():
    tasks = [
        get_stock_quote("AAPL"),
        get_stock_quote("MSFT"),
        get_stock_quote("GOOGL")
    ]
    results = await asyncio.gather(*tasks)
    return results

​
2. Connection Pooling

# Reuse HTTP connections
session = aiohttp.ClientSession(
    connector=aiohttp.TCPConnector(
        limit=100,
        limit_per_host=30
    )
)

​
3. Response Caching

# Cache frequently requested data
@lru_cache(maxsize=1000)
def get_cached_quote(symbol: str, timestamp: int):
    # Cache for 1 minute intervals
    return fetch_quote(symbol)

​
Security Architecture

​
1. API Key Management

# Secure environment variable handling
FMP_API_KEY = os.getenv("FMP_API_KEY", "demo")
if FMP_API_KEY == "demo":
    logger.warning("Using demo API key - limited functionality")

​
2. Input Validation

# Strict parameter validation
def validate_symbol(symbol: str) -> str:
    if not symbol or len(symbol) > 10:
        raise ValueError("Invalid symbol")
    return symbol.upper().strip()

​
3. Rate Limiting

# Protect against abuse
@rate_limit(requests_per_minute=60)
async def protected_endpoint():
    pass

​
Deployment Architecture

​
Development

Local Machine
├── Python 3.11+
├── Virtual Environment
├── Environment Variables
└── Direct API Access

​
Production (Railway)

Railway Platform
├── Automatic Deployment
├── Environment Management
├── Load Balancing
├── SSL Termination
└── Monitoring & Logs

​
Container (Docker)

Docker Container
├── Python Runtime
├── Application Code
├── Dependencies
├── Environment Config
└── Health Checks

​
Monitoring & Observability

​
Logging Strategy

import logging

# Structured logging
logger = logging.getLogger(__name__)
logger.info("Stock quote requested", extra={
    "symbol": symbol,
    "source": "FMP",
    "response_time": response_time
})

​
Health Checks

@app.get("/api/v1/health")
async def health_check():
    return {
        "status": "healthy",
        "tools_available": 32,
        "data_sources": ["YFinance", "FMP"],
        "timestamp": datetime.now().isoformat()
    }

​
Next Steps

Development Setup

Set up local development environment

API Reference

Explore all available endpoints and tools

Deployment Guide

Deploy to production environments

Contributing

Extend and customize the server