financial_mcp_server.pyā¢36.7 kB
#!/usr/bin/env python3
"""
Financial Data MCP Server
A comprehensive Model Context Protocol server for financial data analysis.
Features:
- Portfolio analysis from portfolio.json
- Daily earnings reports
- Analyst upgrades/downgrades
- MACD chart generation
- Stock information retrieval
- Real-time market data (via free yfinance - Yahoo Finance scraper)
Note: Uses yfinance library which is a FREE unofficial Yahoo Finance API scraper.
No API keys required for basic functionality.
"""
import asyncio
import json
import logging
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, Dict, List, Optional, Sequence
import os
import sys
# Third-party imports
import yfinance as yf # FREE Yahoo Finance scraper - no API key needed
import pandas as pd
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
from matplotlib.figure import Figure
import numpy as np
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
# MCP imports
from mcp.server import Server
from mcp.server.models import InitializationOptions
from mcp.types import (
Resource,
Tool,
TextContent,
ImageContent,
EmbeddedResource,
LoggingLevel
)
# Removed unused import or replace with the correct module if needed
#from mcp.server.models.server_capabilities import ServerCapabilities
#from mcp.types import NotificationOptions
import mcp.server.stdio
import mcp.types as types
# Set up logging
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('/tmp/financial_mcp_server.log'),
logging.StreamHandler(sys.stderr)
]
)
logger = logging.getLogger(__name__)
# Configuration
#CHARTS_DIR = Path.home() / "financial_charts"
CHARTS_DIR = Path(__file__).parent / "financial_charts"
CHARTS_DIR.mkdir(exist_ok=True)
#PORTFOLIO_FILE = Path.home() / "portfolio.json" # Default portfolio location
PORTFOLIO_FILE = Path(__file__).parent / "portfolio.json" # Default portfolio location
class NotificationOptions:
"""Options for server notifications."""
# This class can be extended with specific notification options if needed
def __init__(self):
self.resources_changed = False
self.tools_changed = False
self.prompts_changed = False
class FinancialDataServer:
"""Main server class for financial data operations."""
def __init__(self):
self.server = Server("financial-data-server")
self.session = self._create_session()
# Note: yfinance is FREE - no API key required!
# Optional: Alpha Vantage for additional data sources
self.alpha_vantage_key = os.getenv('ALPHA_VANTAGE_API_KEY', 'demo')
self.portfolio_file = PORTFOLIO_FILE
# Register handlers
self._register_handlers()
def _create_session(self) -> requests.Session:
"""Create a requests session with retry strategy."""
session = requests.Session()
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)
return session
def _register_handlers(self):
"""Register all MCP handlers."""
@self.server.list_resources()
async def handle_list_resources() -> list[Resource]:
"""List available resources."""
return [
Resource(
uri="financial://portfolio/overview",
name="Portfolio Overview",
description="Complete portfolio analysis from portfolio.json",
mimeType="application/json",
),
Resource(
uri="financial://earnings/today",
name="Today's Earnings",
description="Companies reporting earnings today",
mimeType="application/json",
),
Resource(
uri="financial://market/status",
name="Market Status",
description="Current market status and overview",
mimeType="application/json",
),
]
def make_json_serializable(obj):
"""
Convert non-JSON serializable objects to JSON serializable format.
This fixes the 'Object of type bool is not JSON serializable' error.
"""
if obj is None:
return None
elif isinstance(obj, (bool, int, float, str)):
return obj
elif isinstance(obj, (datetime, date)):
return obj.isoformat()
elif isinstance(obj, np.generic):
# Convert numpy scalars to Python types
if isinstance(obj, np.bool_):
return bool(obj)
elif isinstance(obj, np.integer):
return int(obj)
elif isinstance(obj, np.floating):
return float(obj)
else:
return obj.item()
elif isinstance(obj, np.ndarray):
return obj.tolist()
elif isinstance(obj, (pd.Series, pd.DataFrame)):
return obj.to_dict()
elif pd.isna(obj):
return None
elif isinstance(obj, dict):
return {k: make_json_serializable(v) for k, v in obj.items()}
elif isinstance(obj, (list, tuple)):
return [make_json_serializable(item) for item in obj]
else:
# For any other type, convert to string
return str(obj)
@self.server.read_resource()
async def handle_read_resource(uri: str) -> str:
"""Read a specific resource."""
if uri == "financial://portfolio/overview":
portfolio = await self._get_portfolio_overview()
return json.dumps(portfolio, indent=2)
elif uri == "financial://earnings/today":
earnings = await self._get_earnings_today()
return json.dumps(earnings, indent=2)
elif uri == "financial://market/status":
status = await self._get_market_status()
return json.dumps(status, indent=2)
else:
raise ValueError(f"Unknown resource: {uri}")
@self.server.list_tools()
async def handle_list_tools() -> list[Tool]:
"""List available tools."""
return [
Tool(
name="load_portfolio",
description="Load and analyze portfolio from portfolio.json file",
inputSchema={
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "Path to portfolio.json file (default: ~/portfolio.json)"
}
}
},
),
Tool(
name="analyze_portfolio",
description="Get detailed analysis of a specific portfolio from portfolio.json",
inputSchema={
"type": "object",
"properties": {
"portfolio_name": {
"type": "string",
"description": "Name of portfolio to analyze (e.g., 'tech_stocks', 'dividend_portfolio')"
},
"file_path": {
"type": "string",
"description": "Path to portfolio.json file (default: ~/portfolio.json)"
}
},
"required": ["portfolio_name"]
},
),
Tool(
name="portfolio_performance",
description="Get performance metrics for all portfolios or a specific one",
inputSchema={
"type": "object",
"properties": {
"portfolio_name": {
"type": "string",
"description": "Specific portfolio name (optional, analyzes all if not provided)"
},
"period": {
"type": "string",
"description": "Time period for performance (1d, 5d, 1mo, 3mo, 6mo, 1y)",
"default": "1mo"
}
}
},
),
Tool(
name="get_stock_info",
description="Get comprehensive stock information including price, fundamentals, and company details",
inputSchema={
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "Stock ticker symbol (e.g., AAPL, GOOGL)"
}
},
"required": ["symbol"]
},
),
Tool(
name="get_earnings_calendar",
description="Get upcoming earnings announcements for the next few days",
inputSchema={
"type": "object",
"properties": {
"days_ahead": {
"type": "integer",
"description": "Number of days to look ahead (default: 7)",
"default": 7
}
}
},
),
Tool(
name="get_analyst_changes",
description="Get recent analyst upgrades and downgrades",
inputSchema={
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "Stock ticker symbol (optional, if not provided returns general market changes)"
},
"days_back": {
"type": "integer",
"description": "Number of days to look back (default: 7)",
"default": 7
}
}
},
),
Tool(
name="generate_macd_chart",
description="Generate MACD technical analysis chart for a stock",
inputSchema={
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "Stock ticker symbol"
},
"period": {
"type": "string",
"description": "Time period for chart (1mo, 3mo, 6mo, 1y, 2y)",
"default": "6mo"
},
"save_path": {
"type": "string",
"description": "Optional custom save path (defaults to ~/financial_charts/)",
"default": ""
}
},
"required": ["symbol"]
},
),
Tool(
name="get_market_overview",
description="Get general market overview including major indices",
inputSchema={
"type": "object",
"properties": {}
},
),
]
@self.server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
"""Handle tool calls."""
try:
if name == "load_portfolio":
file_path = arguments.get("file_path", str(self.portfolio_file))
result = await self._load_portfolio(file_path)
return [types.TextContent(type="text", text=json.dumps(result, indent=2))]
elif name == "analyze_portfolio":
portfolio_name = arguments["portfolio_name"]
file_path = arguments.get("file_path", str(self.portfolio_file))
result = await self._analyze_portfolio(portfolio_name, file_path)
return [types.TextContent(type="text", text=json.dumps(result, indent=2))]
elif name == "portfolio_performance":
portfolio_name = arguments.get("portfolio_name")
period = arguments.get("period", "1mo")
result = await self._portfolio_performance(portfolio_name, period)
return [types.TextContent(type="text", text=json.dumps(result, indent=2))]
elif name == "get_stock_info":
result = await self._get_stock_info(arguments["symbol"])
return [types.TextContent(type="text", text=json.dumps(result, indent=2))]
elif name == "get_earnings_calendar":
days_ahead = arguments.get("days_ahead", 7)
result = await self._get_earnings_calendar(days_ahead)
return [types.TextContent(type="text", text=json.dumps(result, indent=2))]
elif name == "get_analyst_changes":
symbol = arguments.get("symbol")
days_back = arguments.get("days_back", 7)
result = await self._get_analyst_changes(symbol, days_back)
return [types.TextContent(type="text", text=json.dumps(result, indent=2))]
elif name == "generate_macd_chart":
symbol = arguments["symbol"]
period = arguments.get("period", "6mo")
save_path = arguments.get("save_path", "")
result = await self._generate_macd_chart(symbol, period, save_path)
return [types.TextContent(type="text", text=json.dumps(result, indent=2))]
elif name == "get_market_overview":
result = await self._get_market_overview()
return [types.TextContent(type="text", text=json.dumps(result, indent=2))]
else:
raise ValueError(f"Unknown tool: {name}")
except Exception as e:
logger.error(f"Error in tool {name}: {str(e)}")
return [types.TextContent(type="text", text=f"Error: {str(e)}")]
async def _load_portfolio(self, file_path: str) -> Dict[str, Any]:
"""Load portfolio data from JSON file."""
try:
portfolio_path = Path(file_path)
if not portfolio_path.exists():
return {
"error": f"Portfolio file not found: {file_path}",
"suggestion": "Create a portfolio.json file with your stock portfolios",
"example_structure": {
"tech_stocks": {
"portfolio": "Technology Giants",
"stock_list": ["AAPL", "GOOGL", "MSFT", "AMZN", "META"]
}
}
}
with open(portfolio_path, 'r') as f:
portfolio_data = json.load(f)
# Get basic info for each portfolio
portfolio_summary = {}
for portfolio_id, portfolio_info in portfolio_data.items():
stock_list = portfolio_info.get('stock_list', [])
portfolio_summary[portfolio_id] = {
"display_name": portfolio_info.get('portfolio', portfolio_id),
"stock_count": len(stock_list),
"stocks": stock_list
}
return {
"portfolios_loaded": len(portfolio_data),
"portfolio_summary": portfolio_summary,
"file_path": file_path,
"timestamp": datetime.now().isoformat()
}
except Exception as e:
logger.error(f"Error loading portfolio: {str(e)}")
return {
"error": f"Failed to load portfolio: {str(e)}",
"file_path": file_path,
"timestamp": datetime.now().isoformat()
}
async def _analyze_portfolio(self, portfolio_name: str, file_path: str) -> Dict[str, Any]:
"""Analyze a specific portfolio in detail."""
try:
portfolio_path = Path(file_path)
if not portfolio_path.exists():
return {"error": f"Portfolio file not found: {file_path}"}
with open(portfolio_path, 'r') as f:
portfolio_data = json.load(f)
if portfolio_name not in portfolio_data:
available = list(portfolio_data.keys())
return {
"error": f"Portfolio '{portfolio_name}' not found",
"available_portfolios": available
}
portfolio_info = portfolio_data[portfolio_name]
stock_list = portfolio_info.get('stock_list', [])
# Analyze each stock in the portfolio
stock_analysis = {}
portfolio_value = 0
portfolio_change = 0
successful_stocks = 0
for symbol in stock_list:
try:
stock_data = await self._get_stock_info(symbol)
if 'error' not in stock_data:
stock_analysis[symbol] = stock_data
# Assume 1 share for simplicity (real portfolios would have quantities)
portfolio_value += stock_data.get('current_price', 0)
portfolio_change += stock_data.get('change', 0)
successful_stocks += 1
except:
stock_analysis[symbol] = {"error": "Could not fetch data"}
return {
"portfolio_name": portfolio_name,
"display_name": portfolio_info.get('portfolio', portfolio_name),
"total_stocks": len(stock_list),
"successful_analysis": successful_stocks,
"portfolio_metrics": {
"total_value_1_share_each": round(portfolio_value, 2),
"total_change_1_share_each": round(portfolio_change, 2),
"avg_change_percent": round(portfolio_change / max(successful_stocks, 1), 2)
},
"stock_analysis": stock_analysis,
"timestamp": datetime.now().isoformat()
}
except Exception as e:
logger.error(f"Error analyzing portfolio {portfolio_name}: {str(e)}")
return {
"error": f"Failed to analyze portfolio: {str(e)}",
"portfolio_name": portfolio_name,
"timestamp": datetime.now().isoformat()
}
async def _portfolio_performance(self, portfolio_name: Optional[str], period: str) -> Dict[str, Any]:
"""Get performance metrics for portfolios."""
try:
if not self.portfolio_file.exists():
return {"error": f"Portfolio file not found: {self.portfolio_file}"}
with open(self.portfolio_file, 'r') as f:
portfolio_data = json.load(f)
if portfolio_name and portfolio_name not in portfolio_data:
return {
"error": f"Portfolio '{portfolio_name}' not found",
"available_portfolios": list(portfolio_data.keys())
}
# Analyze specified portfolio or all portfolios
portfolios_to_analyze = {portfolio_name: portfolio_data[portfolio_name]} if portfolio_name else portfolio_data
performance_data = {}
for port_name, port_info in portfolios_to_analyze.items():
stock_list = port_info.get('stock_list', [])
portfolio_performance = []
for symbol in stock_list:
try:
# Get historical data for performance calculation
stock = yf.Ticker(symbol)
hist = stock.history(period=period)
if not hist.empty and len(hist) > 1:
start_price = hist['Close'].iloc[0]
end_price = hist['Close'].iloc[-1]
performance = ((end_price - start_price) / start_price) * 100
portfolio_performance.append({
"symbol": symbol,
"start_price": round(float(start_price), 2),
"current_price": round(float(end_price), 2),
"performance_percent": round(float(performance), 2)
})
except:
portfolio_performance.append({
"symbol": symbol,
"error": "Could not calculate performance"
})
# Calculate portfolio averages
valid_performances = [s['performance_percent'] for s in portfolio_performance if 'performance_percent' in s]
avg_performance = sum(valid_performances) / len(valid_performances) if valid_performances else 0
performance_data[port_name] = {
"display_name": port_info.get('portfolio', port_name),
"period": period,
"stock_performances": portfolio_performance,
"portfolio_avg_performance": round(avg_performance, 2),
"best_performer": max(valid_performances) if valid_performances else None,
"worst_performer": min(valid_performances) if valid_performances else None
}
return {
"performance_analysis": performance_data,
"period": period,
"analysis_date": datetime.now().isoformat()
}
except Exception as e:
logger.error(f"Error calculating portfolio performance: {str(e)}")
return {
"error": f"Failed to calculate performance: {str(e)}",
"timestamp": datetime.now().isoformat()
}
async def _get_portfolio_overview(self) -> Dict[str, Any]:
"""Get portfolio overview (helper for resources)."""
return await self._load_portfolio(str(self.portfolio_file))
async def _get_stock_info(self, symbol: str) -> Dict[str, Any]:
"""Get comprehensive stock information."""
try:
stock = yf.Ticker(symbol.upper())
info = stock.info
hist = stock.history(period="5d")
if hist.empty:
raise ValueError(f"No data found for symbol {symbol}")
current_price = hist['Close'].iloc[-1]
prev_close = info.get('previousClose', hist['Close'].iloc[-2] if len(hist) > 1 else current_price)
change = current_price - prev_close
change_percent = (change / prev_close) * 100
return {
"symbol": symbol.upper(),
"company_name": info.get('longName', 'N/A'),
"current_price": round(float(current_price), 2),
"previous_close": round(float(prev_close), 2),
"change": round(float(change), 2),
"change_percent": round(float(change_percent), 2),
"volume": info.get('volume', 'N/A'),
"market_cap": info.get('marketCap', 'N/A'),
"pe_ratio": info.get('trailingPE', 'N/A'),
"dividend_yield": info.get('dividendYield', 'N/A'),
"52_week_high": info.get('fiftyTwoWeekHigh', 'N/A'),
"52_week_low": info.get('fiftyTwoWeekLow', 'N/A'),
"sector": info.get('sector', 'N/A'),
"industry": info.get('industry', 'N/A'),
"business_summary": info.get('longBusinessSummary', 'N/A')[:500] + "..." if info.get('longBusinessSummary') else 'N/A',
"timestamp": datetime.now().isoformat()
}
except Exception as e:
logger.error(f"Error getting stock info for {symbol}: {str(e)}")
raise ValueError(f"Could not retrieve stock information for {symbol}: {str(e)}")
async def _get_earnings_calendar(self, days_ahead: int = 7) -> Dict[str, Any]:
"""Get upcoming earnings announcements."""
try:
# Since yfinance doesn't have a direct earnings calendar, we'll use a different approach
# This is a simplified version - in practice, you might want to use a dedicated financial API
# Get earnings for popular stocks as an example
popular_stocks = ['AAPL', 'GOOGL', 'MSFT', 'AMZN', 'TSLA', 'META', 'NVDA', 'NFLX']
earnings_data = []
for symbol in popular_stocks:
try:
stock = yf.Ticker(symbol)
calendar = stock.calendar
if calendar is not None and not calendar.empty:
for date, data in calendar.iterrows():
earnings_data.append({
"symbol": symbol,
"date": date.strftime("%Y-%m-%d") if hasattr(date, 'strftime') else str(date),
"earnings_data": data.to_dict() if hasattr(data, 'to_dict') else str(data)
})
except:
continue
return {
"earnings_calendar": earnings_data,
"days_ahead": days_ahead,
"timestamp": datetime.now().isoformat(),
"note": "This is a sample implementation. For comprehensive earnings data, consider using a dedicated financial data API."
}
except Exception as e:
logger.error(f"Error getting earnings calendar: {str(e)}")
return {
"error": str(e),
"earnings_calendar": [],
"timestamp": datetime.now().isoformat()
}
async def _get_analyst_changes(self, symbol: Optional[str] = None, days_back: int = 7) -> Dict[str, Any]:
"""Get recent analyst upgrades and downgrades."""
try:
changes = []
if symbol:
# Get analyst recommendations for specific stock
stock = yf.Ticker(symbol.upper())
recommendations = stock.recommendations
if recommendations is not None and not recommendations.empty:
# Get recent recommendations
recent_date = datetime.now() - timedelta(days=days_back)
recent_recs = recommendations[recommendations.index >= recent_date]
for date, rec in recent_recs.iterrows():
changes.append({
"symbol": symbol.upper(),
"date": date.strftime("%Y-%m-%d"),
"firm": rec.get('Firm', 'N/A'),
"to_grade": rec.get('To Grade', 'N/A'),
"from_grade": rec.get('From Grade', 'N/A'),
"action": rec.get('Action', 'N/A')
})
return {
"analyst_changes": changes,
"symbol": symbol,
"days_back": days_back,
"timestamp": datetime.now().isoformat(),
"note": "Analyst data availability varies by stock and timeframe."
}
except Exception as e:
logger.error(f"Error getting analyst changes: {str(e)}")
return {
"error": str(e),
"analyst_changes": [],
"timestamp": datetime.now().isoformat()
}
async def _generate_macd_chart(self, symbol: str, period: str = "6mo", save_path: str = "") -> Dict[str, Any]:
"""Generate MACD technical analysis chart."""
try:
# Get stock data
stock = yf.Ticker(symbol.upper())
hist = stock.history(period=period)
if hist.empty:
raise ValueError(f"No data found for symbol {symbol}")
# Calculate MACD
close_prices = hist['Close']
# MACD parameters
exp1 = close_prices.ewm(span=12).mean()
exp2 = close_prices.ewm(span=26).mean()
macd_line = exp1 - exp2
signal_line = macd_line.ewm(span=9).mean()
histogram = macd_line - signal_line
# Create the chart
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 10), sharex=True)
# Price chart
ax1.plot(hist.index, close_prices, label=f'{symbol.upper()} Price', linewidth=2)
ax1.set_title(f'{symbol.upper()} Stock Price and MACD Analysis', fontsize=16, fontweight='bold')
ax1.set_ylabel('Price ($)', fontsize=12)
ax1.grid(True, alpha=0.3)
ax1.legend()
# MACD chart
ax2.plot(hist.index, macd_line, label='MACD', color='blue', linewidth=2)
ax2.plot(hist.index, signal_line, label='Signal', color='red', linewidth=2)
ax2.bar(hist.index, histogram, label='Histogram', alpha=0.3, color='gray')
ax2.axhline(y=0, color='black', linestyle='-', alpha=0.5)
ax2.set_title('MACD Indicator', fontsize=14)
ax2.set_ylabel('MACD', fontsize=12)
ax2.set_xlabel('Date', fontsize=12)
ax2.grid(True, alpha=0.3)
ax2.legend()
# Format x-axis
ax2.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d'))
ax2.xaxis.set_major_locator(mdates.MonthLocator())
plt.xticks(rotation=45)
plt.tight_layout()
# Save the chart
if save_path:
save_dir = Path(save_path)
else:
save_dir = CHARTS_DIR
save_dir.mkdir(exist_ok=True, parents=True)
filename = f"{symbol.upper()}_MACD_{period}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.png"
full_path = save_dir / filename
plt.savefig(full_path, dpi=300, bbox_inches='tight')
plt.close()
# Calculate current MACD values
current_macd = macd_line.iloc[-1]
current_signal = signal_line.iloc[-1]
current_histogram = histogram.iloc[-1]
return {
"symbol": symbol.upper(),
"chart_saved": str(full_path),
"period": period,
"current_values": {
"macd": round(float(current_macd), 4),
"signal": round(float(current_signal), 4),
"histogram": round(float(current_histogram), 4),
"macd_above_signal": bool(current_macd > current_signal) # Convert to Python bool
},
"analysis": {
"trend": "Bullish" if current_macd > current_signal else "Bearish",
"momentum": "Increasing" if current_histogram > 0 else "Decreasing"
},
"timestamp": datetime.now().isoformat()
}
except Exception as e:
logger.error(f"Error generating MACD chart for {symbol}: {str(e)}")
raise ValueError(f"Could not generate MACD chart for {symbol}: {str(e)}")
async def _get_market_overview(self) -> Dict[str, Any]:
"""Get general market overview."""
try:
indices = {
"S&P 500": "^GSPC",
"Dow Jones": "^DJI",
"NASDAQ": "^IXIC",
"Russell 2000": "^RUT",
"VIX": "^VIX"
}
market_data = {}
for name, symbol in indices.items():
try:
ticker = yf.Ticker(symbol)
hist = ticker.history(period="2d")
if not hist.empty:
current = hist['Close'].iloc[-1]
previous = hist['Close'].iloc[-2] if len(hist) > 1 else current
change = current - previous
change_percent = (change / previous) * 100
market_data[name] = {
"symbol": symbol,
"current": round(float(current), 2),
"change": round(float(change), 2),
"change_percent": round(float(change_percent), 2)
}
except:
market_data[name] = {"error": "Data not available"}
return {
"market_overview": market_data,
"timestamp": datetime.now().isoformat(),
"market_status": "Market data retrieved successfully"
}
except Exception as e:
logger.error(f"Error getting market overview: {str(e)}")
return {
"error": str(e),
"timestamp": datetime.now().isoformat()
}
async def _get_earnings_today(self) -> Dict[str, Any]:
"""Get today's earnings (helper for resources)."""
return await self._get_earnings_calendar(1)
async def _get_market_status(self) -> Dict[str, Any]:
"""Get market status (helper for resources)."""
return await self._get_market_overview()
async def run(self):
"""Run the MCP server."""
async with mcp.server.stdio.stdio_server() as streams:
await self.server.run(
streams[0], streams[1], InitializationOptions(
server_name="financial_data_server",
server_version="1.0.0",
capabilities=self.server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities=None
),
)
)
async def main():
"""Main entry point."""
logger.info("=== Financial MCP Server Starting ===")
logger.info(f"Python path: {sys.executable}")
logger.info(f"Working directory: {os.getcwd()}")
logger.info(f"Command line args: {sys.argv}")
server = FinancialDataServer()
await server.run()
if __name__ == "__main__":
asyncio.run(main())