test_server.py

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