stock_ticker_mcp.py•6.13 kB
#!/usr/bin/env python3
"""
Stock Ticker MCP Server
A Model Context Protocol server that provides stock market data using Alpha Vantage API.
This server exposes tools for getting current stock prices and historical price data.
"""
import os
import httpx
import logging
import sys
from typing import Any
#from fastmcp import FastMCP
from mcp.server.fastmcp import FastMCP
from datetime import datetime
mcp = FastMCP("StockTicker")
# Configure logging to stderr
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
stream=sys.stderr
)
logger = logging.getLogger("stock-ticker-server")
#ALPHA_VANTAGE_API_KEY = os.getenv("ALPHA_VANTAGE_API_KEY")
ALPHA_VANTAGE_API_KEY = "OJ1H3DHE8BYFYBWH" #duck.garby
ALPHA_VANTAGE_BASE_URL = "https://www.alphavantage.co/query"
@mcp.tool()
def get_current_price(symbol: str) -> dict[str, Any]:
"""
Get the current stock price for a given symbol.
Args:
symbol: Stock symbol (e.g., "AAPL", "MSFT", "GOOGL")
Returns:
Dictionary containing current stock price information
"""
if not ALPHA_VANTAGE_API_KEY:
return {"error": "Alpha Vantage API key not configured"}
try:
with httpx.Client(timeout=30.0) as client:
response = client.get(
ALPHA_VANTAGE_BASE_URL,
params={
"function": "GLOBAL_QUOTE",
"symbol": symbol.upper(),
"apikey": ALPHA_VANTAGE_API_KEY
}
)
response.raise_for_status()
data = response.json()
if "Error Message" in data:
return {"error": f"Invalid symbol '{symbol}' or API error"}
if "Note" in data:
return {"error": "API call frequency limit reached. Please try again later."}
if "Global Quote" not in data:
return {"error": "No data available for this symbol"}
quote = data["Global Quote"]
return {
"symbol": quote["01. symbol"],
"price": float(quote["05. price"]),
"change": float(quote["09. change"]),
"change_percent": quote["10. change percent"].replace("%", ""),
"open": float(quote["02. open"]),
"high": float(quote["03. high"]),
"low": float(quote["04. low"]),
"volume": int(quote["06. volume"]),
"latest_trading_day": quote["07. latest trading day"],
"previous_close": float(quote["08. previous close"])
}
except httpx.TimeoutException:
return {"error": "Request timed out. Please try again."}
except httpx.HTTPStatusError as e:
return {"error": f"HTTP error: {e.response.status_code}"}
except ValueError as e:
return {"error": f"Data parsing error: {str(e)}"}
except Exception as e:
return {"error": f"Failed to fetch stock data: {str(e)}"}
@mcp.tool()
def get_historical_price(symbol: str, date: str) -> dict[str, Any]:
"""
Get historical stock price for a given symbol on a specific date.
Args:
symbol: Stock symbol (e.g., "AAPL", "MSFT", "GOOGL")
date: Date in YYYY-MM-DD format (e.g., "2024-01-15")
Returns:
Dictionary containing historical stock price information for the specified date
"""
if not ALPHA_VANTAGE_API_KEY:
return {"error": "Alpha Vantage API key not configured"}
# Validate date format
try:
datetime.strptime(date, "%Y-%m-%d")
except ValueError:
return {"error": "Invalid date format. Please use YYYY-MM-DD format (e.g., '2024-01-15')"}
try:
with httpx.Client(timeout=30.0) as client:
response = client.get(
ALPHA_VANTAGE_BASE_URL,
params={
"function": "TIME_SERIES_DAILY",
"symbol": symbol.upper(),
"apikey": ALPHA_VANTAGE_API_KEY,
"outputsize": "full"
}
)
response.raise_for_status()
data = response.json()
if "Error Message" in data:
return {"error": f"Invalid symbol '{symbol}' or API error"}
if "Note" in data:
return {"error": "API call frequency limit reached. Please try again later."}
if "Time Series (Daily)" not in data:
return {"error": "No daily time series data available for this symbol"}
time_series = data["Time Series (Daily)"]
if date not in time_series:
available_dates = sorted(time_series.keys(), reverse=True)[:10]
return {
"error": f"No data available for {date}. This might be a weekend, holiday, or non-trading day.",
"symbol": symbol.upper(),
"requested_date": date,
"recent_available_dates": available_dates
}
day_data = time_series[date]
return {
"symbol": data["Meta Data"]["2. Symbol"],
"date": date,
"open": float(day_data["1. open"]),
"high": float(day_data["2. high"]),
"low": float(day_data["3. low"]),
"close": float(day_data["4. close"]),
"volume": int(day_data["5. volume"])
}
except httpx.TimeoutException:
return {"error": "Request timed out. Please try again."}
except httpx.HTTPStatusError as e:
return {"error": f"HTTP error: {e.response.status_code}"}
except ValueError as e:
return {"error": f"Data parsing error: {str(e)}"}
except Exception as e:
return {"error": f"Failed to fetch historical data: {str(e)}"}
# === SERVER STARTUP ===
if __name__ == "__main__":
logger.info("Starting Stock Ticker MCP server...")
try:
mcp.run(transport='stdio')
except Exception as e:
logger.error(f"Server error: {e}", exc_info=True)
sys.exit(1)