#!/usr/bin/env python3
"""
Test script for Financial MCP Server
Run this to verify your installation is working correctly
"""
import asyncio
import sys
try:
import yfinance as yf
import pandas as pd
import numpy as np
from mcp.server.fastmcp import FastMCP
print("OK: All required packages installed")
except ImportError as e:
print(f"✗ Missing package: {e}")
print("Please run: pip install -r requirements.txt")
sys.exit(1)
from server import (
get_stock_price,
get_historical_data,
get_options_chain,
calculate_moving_average,
calculate_rsi,
calculate_sharpe_ratio,
compare_stocks,
get_company_news,
get_dividends,
)
async def test_basic_functionality():
"""Test basic market data retrieval"""
print("\nTesting basic functionality...")
try:
ticker = yf.Ticker("AAPL")
info = ticker.info
current_price = info.get("currentPrice", info.get("regularMarketPrice", "N/A"))
if current_price != "N/A":
print(f"OK: Successfully retrieved AAPL price: ${current_price}")
else:
print("FAIL: Could not retrieve price data")
hist = ticker.history(period="5d")
if not hist.empty:
print(f"OK: Retrieved {len(hist)} days of historical data")
else:
print("FAIL: Could not retrieve historical data")
except Exception as e:
print(f"FAIL: Error during testing: {e}")
return False
return True
async def test_calculations():
"""Test financial calculations"""
print("\nTesting calculations...")
try:
data = pd.Series([100, 102, 101, 103, 105, 104, 106])
sma = data.rolling(window=3).mean()
print(f"OK: SMA calculation working: {sma.iloc[-1]:.2f}")
delta = data.diff()
gain = (delta.where(delta > 0, 0)).rolling(window=3).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=3).mean()
rs = gain / loss
rsi = 100 - (100 / (1 + rs))
print(f"OK: RSI calculation working: {rsi.iloc[-1]:.2f}")
except Exception as e:
print(f"FAIL: Calculation error: {e}")
return False
return True
async def test_mcp_tools():
"""Smoke test MCP tool functions"""
print("\nTesting MCP tools...")
ok = True
async def check(name, coro):
nonlocal ok
try:
res = await coro
if isinstance(res, dict) and ("error" not in res):
print(f"OK: {name}")
else:
msg = res.get("error", "bad result") if isinstance(res, dict) else "bad result"
print(f"FAIL: {name}: {msg}")
ok = False
except Exception as e:
print(f"FAIL: {name}: {e}")
ok = False
await check("get_stock_price", get_stock_price("AAPL"))
await check("get_historical_data", get_historical_data("AAPL", period="1mo", interval="1d"))
await check("get_options_chain", get_options_chain("AAPL", sort_by="openInterest", limit=5))
await check("calculate_moving_average", calculate_moving_average("AAPL", period=20, ma_type="SMA"))
await check("calculate_rsi", calculate_rsi("AAPL", period=14))
await check("calculate_sharpe_ratio", calculate_sharpe_ratio("AAPL", period="1y", risk_free_rate=0.05))
await check("compare_stocks", compare_stocks(["AAPL", "MSFT"], metric="volatility"))
await check("get_company_news", get_company_news("AAPL", limit=2))
await check("get_dividends", get_dividends("KO", period="5y"))
return ok
def main():
print("Financial MCP Server Test Suite")
print("=" * 40)
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
basic_ok = loop.run_until_complete(test_basic_functionality())
calc_ok = loop.run_until_complete(test_calculations())
mcp_ok = loop.run_until_complete(test_mcp_tools())
print("\n" + "=" * 40)
if basic_ok and calc_ok and mcp_ok:
print("OK: All tests passed! Your server is ready to use.")
print("\nTo use with Claude Desktop:")
print("1. Add the server to your Claude Desktop config")
print("2. Restart Claude Desktop")
print("3. Look for 'financial-data' in the MCP tools")
else:
print("FAIL: Some tests failed. Please check the errors above.")
loop.close()
if __name__ == "__main__":
main()
