server.py•14.4 kB
import json
import logging
import os
import pandas as pd
import yfinance as yf
from dotenv import load_dotenv
from fastmcp import FastMCP
load_dotenv()
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger("YahooFinanceMCP")
mcp = FastMCP("Yahoo Finance MCP Server", version="0.1.0")
@mcp.tool
def get_historical_stock_prices(
ticker: str, period: str = "1mo", interval: str = "1d"
) -> str:
"""
Get historical stock prices for a given ticker symbol from yahoo finance.
Args:
ticker: The ticker symbol (e.g. "AAPL")
period: Valid periods: 1d,5d,1mo,3mo,6mo,1y,2y,5y,10y,ytd,max (default: "1mo")
interval: Valid intervals: 1m,2m,5m,15m,30m,60m,90m,1h,1d,5d,1wk,1mo,3mo (default: "1d")
Returns:
JSON string with historical price data including Date, Open, High, Low, Close, Volume, Adj Close
"""
if not ticker:
return "Error: Ticker symbol is required"
try:
company = yf.Ticker(ticker)
try:
if company.info.get("symbol") is None:
return f"Error: Company ticker {ticker} not found"
except:
return f"Error: Company ticker {ticker} not found"
hist_data = company.history(period=period, interval=interval)
if hist_data.empty:
return f"Error: No historical data found for {ticker}"
hist_data = hist_data.reset_index(names="Date")
return hist_data.to_json(orient="records", date_format="iso")
except Exception as e:
logger.error(f"Error getting historical stock prices for {ticker}: {e}")
return f"Error getting historical data: {str(e)}"
@mcp.tool
def get_stock_info(ticker: str) -> str:
"""
Get comprehensive stock information for a given ticker symbol.
Args:
ticker: The ticker symbol (e.g. "AAPL")
Returns:
JSON string with stock information including price, company data, financial metrics
"""
if not ticker:
return "Error: Ticker symbol is required"
try:
company = yf.Ticker(ticker)
info = company.info
if not info or info.get("symbol") is None:
return f"Error: Company ticker {ticker} not found"
return json.dumps(info)
except Exception as e:
logger.error(f"Error getting stock information for {ticker}: {e}")
return f"Error getting stock info: {str(e)}"
@mcp.tool
def get_yahoo_finance_news(ticker: str) -> str:
"""
Get latest news for a given ticker symbol from yahoo finance.
Args:
ticker: The ticker symbol (e.g. "AAPL")
Returns:
Formatted news articles with title, summary, description, and URL
"""
if not ticker:
return "Error: Ticker symbol is required"
try:
company = yf.Ticker(ticker)
try:
if company.info.get("symbol") is None:
return f"Error: Company ticker {ticker} not found"
except:
return f"Error: Company ticker {ticker} not found"
news = company.news
if not news:
return f"No news found for {ticker}"
news_list = []
for article in news:
if article.get("content", {}).get("contentType", "") == "STORY":
title = article.get("content", {}).get("title", "")
summary = article.get("content", {}).get("summary", "")
description = article.get("content", {}).get("description", "")
url = article.get("content", {}).get("canonicalUrl", {}).get("url", "")
news_list.append(
f"Title: {title}\nSummary: {summary}\nDescription: {description}\nURL: {url}"
)
if not news_list:
return f"No news articles found for {ticker}"
return "\n\n".join(news_list)
except Exception as e:
logger.error(f"Error getting news for {ticker}: {e}")
return f"Error getting news: {str(e)}"
@mcp.tool
def get_stock_actions(ticker: str) -> str:
"""
Get stock dividends and stock splits for a given ticker symbol.
Args:
ticker: The ticker symbol (e.g. "AAPL")
Returns:
JSON string with stock actions (dividends and splits) data
"""
if not ticker:
return "Error: Ticker symbol is required"
try:
company = yf.Ticker(ticker)
actions_df = company.actions
if actions_df.empty:
return f"No stock actions found for {ticker}"
actions_df = actions_df.reset_index(names="Date")
return actions_df.to_json(orient="records", date_format="iso")
except Exception as e:
logger.error(f"Error getting stock actions for {ticker}: {e}")
return f"Error getting stock actions: {str(e)}"
@mcp.tool
def get_financial_statement(ticker: str, financial_type: str) -> str:
"""
Get financial statement for a given ticker symbol.
Args:
ticker: The ticker symbol (e.g. "AAPL")
financial_type: Type of financial statement (income_stmt, quarterly_income_stmt, balance_sheet, quarterly_balance_sheet, cashflow, quarterly_cashflow)
Returns:
JSON string with financial statement data
"""
if not ticker:
return "Error: Ticker symbol is required"
valid_financial_types = [
"income_stmt",
"quarterly_income_stmt",
"balance_sheet",
"quarterly_balance_sheet",
"cashflow",
"quarterly_cashflow",
]
if financial_type not in valid_financial_types:
return f"Error: Invalid financial type. Valid types: {valid_financial_types}"
try:
company = yf.Ticker(ticker)
try:
if company.info.get("symbol") is None:
return f"Error: Company ticker {ticker} not found"
except:
return f"Error: Company ticker {ticker} not found"
if financial_type == "income_stmt":
financial_statement = company.income_stmt
elif financial_type == "quarterly_income_stmt":
financial_statement = company.quarterly_income_stmt
elif financial_type == "balance_sheet":
financial_statement = company.balance_sheet
elif financial_type == "quarterly_balance_sheet":
financial_statement = company.quarterly_balance_sheet
elif financial_type == "cashflow":
financial_statement = company.cashflow
elif financial_type == "quarterly_cashflow":
financial_statement = company.quarterly_cashflow
if financial_statement.empty:
return f"No {financial_type} data found for {ticker}"
result = []
for column in financial_statement.columns:
if isinstance(column, pd.Timestamp):
date_str = column.strftime("%Y-%m-%d")
else:
date_str = str(column)
date_obj = {"date": date_str}
for index, value in financial_statement[column].items():
date_obj[index] = None if pd.isna(value) else value
result.append(date_obj)
return json.dumps(result)
except Exception as e:
logger.error(f"Error getting financial statement for {ticker}: {e}")
return f"Error getting financial statement: {str(e)}"
@mcp.tool
def get_holder_info(ticker: str, holder_type: str) -> str:
"""
Get holder information for a given ticker symbol.
Args:
ticker: The ticker symbol (e.g. "AAPL")
holder_type: Type of holder info (major_holders, institutional_holders, mutualfund_holders, insider_transactions, insider_purchases, insider_roster_holders)
Returns:
JSON string with holder information
"""
if not ticker:
return "Error: Ticker symbol is required"
valid_holder_types = [
"major_holders",
"institutional_holders",
"mutualfund_holders",
"insider_transactions",
"insider_purchases",
"insider_roster_holders",
]
if holder_type not in valid_holder_types:
return f"Error: Invalid holder type. Valid types: {valid_holder_types}"
try:
company = yf.Ticker(ticker)
try:
if company.info.get("symbol") is None:
return f"Error: Company ticker {ticker} not found"
except:
return f"Error: Company ticker {ticker} not found"
if holder_type == "major_holders":
data = company.major_holders.reset_index(names="metric")
elif holder_type == "institutional_holders":
data = company.institutional_holders
elif holder_type == "mutualfund_holders":
data = company.mutualfund_holders
elif holder_type == "insider_transactions":
data = company.insider_transactions
elif holder_type == "insider_purchases":
data = company.insider_purchases
elif holder_type == "insider_roster_holders":
data = company.insider_roster_holders
if data.empty:
return f"No {holder_type} data found for {ticker}"
return data.to_json(orient="records", date_format="iso")
except Exception as e:
logger.error(f"Error getting holder info for {ticker}: {e}")
return f"Error getting holder info: {str(e)}"
@mcp.tool
def get_option_expiration_dates(ticker: str) -> str:
"""
Fetch the available options expiration dates for a given ticker symbol.
Args:
ticker: The ticker symbol (e.g. "AAPL")
Returns:
JSON string with available expiration dates
"""
if not ticker:
return "Error: Ticker symbol is required"
try:
company = yf.Ticker(ticker)
try:
if company.info.get("symbol") is None:
return f"Error: Company ticker {ticker} not found"
except:
return f"Error: Company ticker {ticker} not found"
options = company.options
if not options:
return f"No options data found for {ticker}"
return json.dumps(list(options))
except Exception as e:
logger.error(f"Error getting option expiration dates for {ticker}: {e}")
return f"Error getting option dates: {str(e)}"
@mcp.tool
def get_option_chain(ticker: str, expiration_date: str, option_type: str) -> str:
"""
Fetch the option chain for a given ticker symbol, expiration date, and option type.
Args:
ticker: The ticker symbol (e.g. "AAPL")
expiration_date: The expiration date (format: 'YYYY-MM-DD')
option_type: The type of option ('calls' or 'puts')
Returns:
JSON string with option chain data
"""
if not ticker:
return "Error: Ticker symbol is required"
if option_type not in ["calls", "puts"]:
return "Error: Option type must be 'calls' or 'puts'"
try:
company = yf.Ticker(ticker)
try:
if company.info.get("symbol") is None:
return f"Error: Company ticker {ticker} not found"
except:
return f"Error: Company ticker {ticker} not found"
if expiration_date not in company.options:
available_dates = list(company.options)
return f"Error: No options available for {expiration_date}. Available dates: {available_dates}"
option_chain = company.option_chain(expiration_date)
if option_type == "calls":
data = option_chain.calls
else:
data = option_chain.puts
if data.empty:
return f"No {option_type} data found for {ticker} on {expiration_date}"
return data.to_json(orient="records", date_format="iso")
except Exception as e:
logger.error(f"Error getting option chain for {ticker}: {e}")
return f"Error getting option chain: {str(e)}"
@mcp.tool
def get_recommendations(
ticker: str, recommendation_type: str, months_back: int = 12
) -> str:
"""
Get recommendations or upgrades/downgrades for a given ticker symbol.
Args:
ticker: The ticker symbol (e.g. "AAPL")
recommendation_type: Type of recommendation (recommendations, upgrades_downgrades)
months_back: Number of months back for upgrades/downgrades (default: 12)
Returns:
JSON string with recommendation data
"""
if not ticker:
return "Error: Ticker symbol is required"
valid_recommendation_types = ["recommendations", "upgrades_downgrades"]
if recommendation_type not in valid_recommendation_types:
return f"Error: Invalid recommendation type. Valid types: {valid_recommendation_types}"
try:
company = yf.Ticker(ticker)
try:
if company.info.get("symbol") is None:
return f"Error: Company ticker {ticker} not found"
except:
return f"Error: Company ticker {ticker} not found"
if recommendation_type == "recommendations":
data = company.recommendations
if data.empty:
return f"No recommendations data found for {ticker}"
return data.to_json(orient="records")
elif recommendation_type == "upgrades_downgrades":
upgrades_downgrades = company.upgrades_downgrades.reset_index()
if upgrades_downgrades.empty:
return f"No upgrades/downgrades data found for {ticker}"
cutoff_date = pd.Timestamp.now() - pd.DateOffset(months=months_back)
filtered_data = upgrades_downgrades[
upgrades_downgrades["GradeDate"] >= cutoff_date
]
if filtered_data.empty:
return f"No upgrades/downgrades found for {ticker} in the last {months_back} months"
latest_by_firm = filtered_data.drop_duplicates(subset=["Firm"]).sort_values(
"GradeDate", ascending=False
)
return latest_by_firm.to_json(orient="records", date_format="iso")
except Exception as e:
logger.error(f"Error getting recommendations for {ticker}: {e}")
return f"Error getting recommendations: {str(e)}"
if __name__ == "__main__":
host = os.getenv("HOST", "127.0.0.1")
port = int(os.getenv("PORT", "8000"))
print(f"Starting Yahoo Finance MCP Server locally on {host}:{port}...")
mcp.run(transport="streamable-http", host=host, port=port)