stock_mcp.pyā¢34.5 kB
"""Stock Data MCP Server
An MCP server providing stock market data through Yahoo Finance without requiring an API key.
Uses yfinance library to fetch real-time quotes, historical data, company information, and financials.
"""
import argparse
import json
import os
from datetime import datetime, timedelta
from enum import Enum
from typing import Optional, List, Dict, Any
import yfinance as yf
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field, ConfigDict, field_validator
# Module-level constants
CHARACTER_LIMIT = 25000 # Maximum response size in characters
DEFAULT_HISTORY_PERIOD = "1mo" # Default historical data period
# Initialize MCP server
mcp = FastMCP("stock_mcp")
# ============================================================================
# Enums for input validation
# ============================================================================
class ResponseFormat(str, Enum):
"""Output format for tool responses."""
MARKDOWN = "markdown"
JSON = "json"
class HistoryPeriod(str, Enum):
"""Valid periods for historical data."""
ONE_DAY = "1d"
FIVE_DAYS = "5d"
ONE_MONTH = "1mo"
THREE_MONTHS = "3mo"
SIX_MONTHS = "6mo"
ONE_YEAR = "1y"
TWO_YEARS = "2y"
FIVE_YEARS = "5y"
TEN_YEARS = "10y"
YTD = "ytd"
MAX = "max"
class HistoryInterval(str, Enum):
"""Valid intervals for historical data."""
ONE_MINUTE = "1m"
TWO_MINUTES = "2m"
FIVE_MINUTES = "5m"
FIFTEEN_MINUTES = "15m"
THIRTY_MINUTES = "30m"
SIXTY_MINUTES = "60m"
NINETY_MINUTES = "90m"
ONE_HOUR = "1h"
ONE_DAY = "1d"
FIVE_DAYS = "5d"
ONE_WEEK = "1wk"
ONE_MONTH = "1mo"
THREE_MONTHS = "3mo"
class FinancialStatement(str, Enum):
"""Types of financial statements."""
INCOME_STATEMENT = "income"
BALANCE_SHEET = "balance"
CASH_FLOW = "cashflow"
# ============================================================================
# Pydantic Input Models
# ============================================================================
class StockQuoteInput(BaseModel):
"""Input model for fetching stock quotes."""
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
extra='forbid'
)
symbol: str = Field(
...,
description="Stock ticker symbol (e.g., 'AAPL', 'MSFT', 'GOOGL', 'TSLA')",
min_length=1,
max_length=10
)
response_format: ResponseFormat = Field(
default=ResponseFormat.MARKDOWN,
description="Output format: 'markdown' for human-readable or 'json' for machine-readable"
)
@field_validator('symbol')
@classmethod
def validate_symbol(cls, v: str) -> str:
"""Ensure symbol is uppercase and stripped."""
return v.strip().upper()
class StockInfoInput(BaseModel):
"""Input model for fetching detailed stock information."""
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
extra='forbid'
)
symbol: str = Field(
...,
description="Stock ticker symbol (e.g., 'AAPL', 'MSFT', 'GOOGL')",
min_length=1,
max_length=10
)
response_format: ResponseFormat = Field(
default=ResponseFormat.MARKDOWN,
description="Output format: 'markdown' for human-readable or 'json' for machine-readable"
)
@field_validator('symbol')
@classmethod
def validate_symbol(cls, v: str) -> str:
"""Ensure symbol is uppercase and stripped."""
return v.strip().upper()
class StockHistoryInput(BaseModel):
"""Input model for fetching historical stock data."""
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
extra='forbid'
)
symbol: str = Field(
...,
description="Stock ticker symbol (e.g., 'AAPL', 'MSFT', 'GOOGL')",
min_length=1,
max_length=10
)
period: HistoryPeriod = Field(
default=HistoryPeriod.ONE_MONTH,
description="Time period: '1d', '5d', '1mo', '3mo', '6mo', '1y', '2y', '5y', '10y', 'ytd', 'max'"
)
interval: HistoryInterval = Field(
default=HistoryInterval.ONE_DAY,
description="Data interval: '1m', '2m', '5m', '15m', '30m', '60m', '90m', '1h', '1d', '5d', '1wk', '1mo', '3mo'"
)
response_format: ResponseFormat = Field(
default=ResponseFormat.MARKDOWN,
description="Output format: 'markdown' for human-readable or 'json' for machine-readable"
)
@field_validator('symbol')
@classmethod
def validate_symbol(cls, v: str) -> str:
"""Ensure symbol is uppercase and stripped."""
return v.strip().upper()
class StockFinancialsInput(BaseModel):
"""Input model for fetching financial statements."""
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
extra='forbid'
)
symbol: str = Field(
...,
description="Stock ticker symbol (e.g., 'AAPL', 'MSFT', 'GOOGL')",
min_length=1,
max_length=10
)
statement_type: FinancialStatement = Field(
default=FinancialStatement.INCOME_STATEMENT,
description="Financial statement type: 'income' for income statement, 'balance' for balance sheet, 'cashflow' for cash flow statement"
)
quarterly: bool = Field(
default=False,
description="If True, returns quarterly data; if False, returns annual data"
)
response_format: ResponseFormat = Field(
default=ResponseFormat.MARKDOWN,
description="Output format: 'markdown' for human-readable or 'json' for machine-readable"
)
@field_validator('symbol')
@classmethod
def validate_symbol(cls, v: str) -> str:
"""Ensure symbol is uppercase and stripped."""
return v.strip().upper()
class TickerSearchInput(BaseModel):
"""Input model for searching ticker symbols."""
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
extra='forbid'
)
query: str = Field(
...,
description="Company name or partial name to search for (e.g., 'Apple', 'Microsoft', 'Tesla')",
min_length=1,
max_length=100
)
response_format: ResponseFormat = Field(
default=ResponseFormat.MARKDOWN,
description="Output format: 'markdown' for human-readable or 'json' for machine-readable"
)
# ============================================================================
# Helper Functions
# ============================================================================
def format_currency(value: Optional[float]) -> str:
"""Format a number as currency."""
if value is None or value == 0:
return "N/A"
# Handle billions and millions
if abs(value) >= 1_000_000_000:
return f"${value / 1_000_000_000:.2f}B"
elif abs(value) >= 1_000_000:
return f"${value / 1_000_000:.2f}M"
else:
return f"${value:,.2f}"
def format_number(value: Optional[float]) -> str:
"""Format a large number with K/M/B suffixes."""
if value is None or value == 0:
return "N/A"
if abs(value) >= 1_000_000_000:
return f"{value / 1_000_000_000:.2f}B"
elif abs(value) >= 1_000_000:
return f"{value / 1_000_000:.2f}M"
elif abs(value) >= 1_000:
return f"{value / 1_000:.2f}K"
else:
return f"{value:.2f}"
def format_percentage(value: Optional[float]) -> str:
"""Format a decimal as a percentage."""
if value is None:
return "N/A"
return f"{value * 100:.2f}%"
def safe_get(data: Dict, key: str, default: Any = None) -> Any:
"""Safely get a value from a dictionary."""
return data.get(key, default)
def truncate_if_needed(content: str, limit: int = CHARACTER_LIMIT) -> str:
"""Truncate content if it exceeds the character limit."""
if len(content) <= limit:
return content
truncated = content[:limit]
return f"{truncated}\n\nā ļø **Response truncated** - Content exceeded {limit:,} character limit. Try narrowing your query parameters."
# ============================================================================
# MCP Tools
# ============================================================================
@mcp.tool(
name="get_stock_quote",
annotations={
"title": "Get Stock Quote",
"readOnlyHint": True,
"destructiveHint": False,
"idempotentHint": True,
"openWorldHint": True
}
)
async def get_stock_quote(params: StockQuoteInput) -> str:
"""Get current stock quote with price, volume, and market data.
Fetches real-time stock quote information including current price, day's change,
volume, market cap, and trading range. This is the quickest way to get current
stock pricing information.
Args:
params (StockQuoteInput): Validated input parameters containing:
- symbol (str): Stock ticker symbol (e.g., 'AAPL', 'MSFT')
- response_format (ResponseFormat): Output format ('markdown' or 'json')
Returns:
str: Formatted stock quote data in the requested format, including:
- Current price and day's change/percentage
- Opening, high, low prices
- Previous close
- Volume and average volume
- Market capitalization
- 52-week high/low range
- Company name
Example:
>>> get_stock_quote({"symbol": "AAPL", "response_format": "markdown"})
Returns formatted quote for Apple Inc.
"""
try:
ticker = yf.Ticker(params.symbol)
info = ticker.info
# Check if we got valid data
if not info or 'symbol' not in info:
return json.dumps({
"error": f"Unable to fetch data for symbol '{params.symbol}'. Please verify the ticker symbol is correct.",
"suggestion": "Try using the 'search_ticker' tool to find the correct symbol."
}, indent=2)
# Extract quote data with safe defaults
current_price = safe_get(info, 'currentPrice') or safe_get(info, 'regularMarketPrice')
previous_close = safe_get(info, 'previousClose')
if current_price and previous_close:
change = current_price - previous_close
change_percent = (change / previous_close) * 100
else:
change = None
change_percent = None
quote_data = {
"symbol": params.symbol,
"name": safe_get(info, 'longName', params.symbol),
"current_price": current_price,
"change": change,
"change_percent": change_percent,
"open": safe_get(info, 'open') or safe_get(info, 'regularMarketOpen'),
"high": safe_get(info, 'dayHigh') or safe_get(info, 'regularMarketDayHigh'),
"low": safe_get(info, 'dayLow') or safe_get(info, 'regularMarketDayLow'),
"previous_close": previous_close,
"volume": safe_get(info, 'volume') or safe_get(info, 'regularMarketVolume'),
"average_volume": safe_get(info, 'averageVolume'),
"market_cap": safe_get(info, 'marketCap'),
"fifty_two_week_high": safe_get(info, 'fiftyTwoWeekHigh'),
"fifty_two_week_low": safe_get(info, 'fiftyTwoWeekLow'),
"currency": safe_get(info, 'currency', 'USD')
}
if params.response_format == ResponseFormat.JSON:
return json.dumps(quote_data, indent=2)
else:
# Markdown format
change_indicator = "š“" if change and change < 0 else "š¢" if change and change > 0 else "āŖ"
md = f"""# {quote_data['name']} ({quote_data['symbol']})
## Current Quote
**Price:** {format_currency(current_price)} {quote_data['currency']}
**Change:** {change_indicator} {format_currency(change)} ({change_percent:.2f}% if change_percent else 'N/A')
## Trading Data
- **Open:** {format_currency(quote_data['open'])} {quote_data['currency']}
- **High:** {format_currency(quote_data['high'])} {quote_data['currency']}
- **Low:** {format_currency(quote_data['low'])} {quote_data['currency']}
- **Previous Close:** {format_currency(quote_data['previous_close'])} {quote_data['currency']}
## Volume & Market Cap
- **Volume:** {format_number(quote_data['volume'])}
- **Avg Volume:** {format_number(quote_data['average_volume'])}
- **Market Cap:** {format_currency(quote_data['market_cap'])}
## 52-Week Range
- **High:** {format_currency(quote_data['fifty_two_week_high'])} {quote_data['currency']}
- **Low:** {format_currency(quote_data['fifty_two_week_low'])} {quote_data['currency']}
"""
return truncate_if_needed(md)
except Exception as e:
error_response = {
"error": f"Failed to fetch stock quote for '{params.symbol}'",
"details": str(e),
"suggestion": "Verify the ticker symbol is correct and try again."
}
return json.dumps(error_response, indent=2)
@mcp.tool(
name="get_stock_info",
annotations={
"title": "Get Detailed Stock Information",
"readOnlyHint": True,
"destructiveHint": False,
"idempotentHint": True,
"openWorldHint": True
}
)
async def get_stock_info(params: StockInfoInput) -> str:
"""Get comprehensive company and stock information.
Fetches detailed information about a company including business description,
sector, industry, key executives, financial metrics, analyst recommendations,
and trading information. Use this for in-depth company analysis.
Args:
params (StockInfoInput): Validated input parameters containing:
- symbol (str): Stock ticker symbol
- response_format (ResponseFormat): Output format ('markdown' or 'json')
Returns:
str: Comprehensive company information including:
- Company description and business summary
- Sector and industry classification
- Key financial metrics (P/E, EPS, dividend yield, etc.)
- Analyst recommendations and target price
- Company officers
- Exchange and trading information
Example:
>>> get_stock_info({"symbol": "TSLA", "response_format": "markdown"})
Returns detailed information about Tesla Inc.
"""
try:
ticker = yf.Ticker(params.symbol)
info = ticker.info
if not info or 'symbol' not in info:
return json.dumps({
"error": f"Unable to fetch information for symbol '{params.symbol}'",
"suggestion": "Verify the ticker symbol and try again."
}, indent=2)
company_info = {
"symbol": params.symbol,
"name": safe_get(info, 'longName'),
"description": safe_get(info, 'longBusinessSummary'),
"sector": safe_get(info, 'sector'),
"industry": safe_get(info, 'industry'),
"website": safe_get(info, 'website'),
"employees": safe_get(info, 'fullTimeEmployees'),
"country": safe_get(info, 'country'),
"city": safe_get(info, 'city'),
"exchange": safe_get(info, 'exchange'),
"currency": safe_get(info, 'currency'),
"market_cap": safe_get(info, 'marketCap'),
"pe_ratio": safe_get(info, 'trailingPE'),
"forward_pe": safe_get(info, 'forwardPE'),
"peg_ratio": safe_get(info, 'pegRatio'),
"price_to_book": safe_get(info, 'priceToBook'),
"earnings_per_share": safe_get(info, 'trailingEps'),
"dividend_yield": safe_get(info, 'dividendYield'),
"beta": safe_get(info, 'beta'),
"recommendation": safe_get(info, 'recommendationKey'),
"target_mean_price": safe_get(info, 'targetMeanPrice'),
"number_of_analysts": safe_get(info, 'numberOfAnalystOpinions')
}
if params.response_format == ResponseFormat.JSON:
return json.dumps(company_info, indent=2)
else:
# Markdown format
md = f"""# {company_info['name']} ({params.symbol})
## Company Overview
{company_info['description'] or 'No description available'}
## Classification
- **Sector:** {company_info['sector'] or 'N/A'}
- **Industry:** {company_info['industry'] or 'N/A'}
- **Exchange:** {company_info['exchange'] or 'N/A'}
## Location & Size
- **Headquarters:** {company_info['city'] or 'N/A'}, {company_info['country'] or 'N/A'}
- **Employees:** {format_number(company_info['employees'])}
- **Website:** {company_info['website'] or 'N/A'}
## Financial Metrics
- **Market Cap:** {format_currency(company_info['market_cap'])}
- **P/E Ratio:** {company_info['pe_ratio']:.2f if company_info['pe_ratio'] else 'N/A'}
- **Forward P/E:** {company_info['forward_pe']:.2f if company_info['forward_pe'] else 'N/A'}
- **PEG Ratio:** {company_info['peg_ratio']:.2f if company_info['peg_ratio'] else 'N/A'}
- **Price/Book:** {company_info['price_to_book']:.2f if company_info['price_to_book'] else 'N/A'}
- **EPS:** {format_currency(company_info['earnings_per_share'])}
- **Beta:** {company_info['beta']:.2f if company_info['beta'] else 'N/A'}
- **Dividend Yield:** {format_percentage(company_info['dividend_yield'])}
## Analyst Opinion
- **Recommendation:** {company_info['recommendation'].upper() if company_info['recommendation'] else 'N/A'}
- **Target Price:** {format_currency(company_info['target_mean_price'])} {company_info['currency']}
- **Number of Analysts:** {company_info['number_of_analysts'] or 'N/A'}
"""
return truncate_if_needed(md)
except Exception as e:
error_response = {
"error": f"Failed to fetch stock information for '{params.symbol}'",
"details": str(e),
"suggestion": "Verify the ticker symbol and try again."
}
return json.dumps(error_response, indent=2)
@mcp.tool(
name="get_stock_history",
annotations={
"title": "Get Historical Stock Data",
"readOnlyHint": True,
"destructiveHint": False,
"idempotentHint": True,
"openWorldHint": True
}
)
async def get_stock_history(params: StockHistoryInput) -> str:
"""Get historical price and volume data for technical analysis.
Fetches historical OHLCV (Open, High, Low, Close, Volume) data for a specified
time period and interval. Useful for price trend analysis, charting, and
identifying patterns.
Args:
params (StockHistoryInput): Validated input parameters containing:
- symbol (str): Stock ticker symbol
- period (HistoryPeriod): Time period ('1d', '5d', '1mo', '3mo', '6mo', '1y', '2y', '5y', '10y', 'ytd', 'max')
- interval (HistoryInterval): Data interval ('1m', '5m', '1h', '1d', '1wk', '1mo', etc.)
- response_format (ResponseFormat): Output format ('markdown' or 'json')
Returns:
str: Historical price data including:
- Date/timestamp for each data point
- Open, High, Low, Close prices
- Trading volume
- Summary statistics (min, max, average)
Example:
>>> get_stock_history({"symbol": "NVDA", "period": "3mo", "interval": "1d"})
Returns 3 months of daily price data for NVIDIA
"""
try:
ticker = yf.Ticker(params.symbol)
# Fetch historical data
hist = ticker.history(period=params.period.value, interval=params.interval.value)
if hist.empty:
return json.dumps({
"error": f"No historical data found for '{params.symbol}' with period '{params.period.value}' and interval '{params.interval.value}'",
"suggestion": "Try a different period/interval combination or verify the ticker symbol."
}, indent=2)
# Convert DataFrame to list of dicts
history_data = []
for index, row in hist.iterrows():
history_data.append({
"date": index.strftime("%Y-%m-%d %H:%M:%S") if hasattr(index, 'strftime') else str(index),
"open": round(row['Open'], 2) if 'Open' in row else None,
"high": round(row['High'], 2) if 'High' in row else None,
"low": round(row['Low'], 2) if 'Low' in row else None,
"close": round(row['Close'], 2) if 'Close' in row else None,
"volume": int(row['Volume']) if 'Volume' in row else None
})
# Calculate summary statistics
summary = {
"symbol": params.symbol,
"period": params.period.value,
"interval": params.interval.value,
"data_points": len(history_data),
"start_date": history_data[0]["date"] if history_data else None,
"end_date": history_data[-1]["date"] if history_data else None,
"high": round(hist['High'].max(), 2) if 'High' in hist.columns else None,
"low": round(hist['Low'].min(), 2) if 'Low' in hist.columns else None,
"avg_close": round(hist['Close'].mean(), 2) if 'Close' in hist.columns else None,
"avg_volume": int(hist['Volume'].mean()) if 'Volume' in hist.columns else None
}
if params.response_format == ResponseFormat.JSON:
response = {
"summary": summary,
"history": history_data
}
content = json.dumps(response, indent=2)
return truncate_if_needed(content)
else:
# Markdown format
md = f"""# Historical Data: {params.symbol}
## Summary
- **Period:** {summary['period']}
- **Interval:** {summary['interval']}
- **Data Points:** {summary['data_points']}
- **Date Range:** {summary['start_date']} to {summary['end_date']}
## Statistics
- **Highest Price:** ${summary['high']:.2f}
- **Lowest Price:** ${summary['low']:.2f}
- **Average Close:** ${summary['avg_close']:.2f}
- **Average Volume:** {format_number(summary['avg_volume'])}
## Recent Data (Last 10 entries)
"""
# Show last 10 entries
for entry in history_data[-10:]:
md += f"\n### {entry['date']}\n"
md += f"- **Open:** ${entry['open']:.2f}\n"
md += f"- **High:** ${entry['high']:.2f}\n"
md += f"- **Low:** ${entry['low']:.2f}\n"
md += f"- **Close:** ${entry['close']:.2f}\n"
md += f"- **Volume:** {format_number(entry['volume'])}\n"
if len(history_data) > 10:
md += f"\n*Showing last 10 of {len(history_data)} data points. Use JSON format for complete data.*\n"
return truncate_if_needed(md)
except Exception as e:
error_response = {
"error": f"Failed to fetch historical data for '{params.symbol}'",
"details": str(e),
"suggestion": "Verify the ticker symbol and ensure the period/interval combination is valid."
}
return json.dumps(error_response, indent=2)
@mcp.tool(
name="get_stock_financials",
annotations={
"title": "Get Financial Statements",
"readOnlyHint": True,
"destructiveHint": False,
"idempotentHint": True,
"openWorldHint": True
}
)
async def get_stock_financials(params: StockFinancialsInput) -> str:
"""Get company financial statements for fundamental analysis.
Fetches detailed financial statements including income statements, balance sheets,
and cash flow statements. Choose between annual and quarterly reporting periods.
Essential for fundamental analysis and company valuation.
Args:
params (StockFinancialsInput): Validated input parameters containing:
- symbol (str): Stock ticker symbol
- statement_type (FinancialStatement): Type of statement ('income', 'balance', 'cashflow')
- quarterly (bool): If True, returns quarterly data; if False, returns annual data
- response_format (ResponseFormat): Output format ('markdown' or 'json')
Returns:
str: Financial statement data including:
- Income Statement: Revenue, expenses, net income, EPS, etc.
- Balance Sheet: Assets, liabilities, equity, etc.
- Cash Flow: Operating, investing, financing activities, etc.
Example:
>>> get_stock_financials({"symbol": "AAPL", "statement_type": "income", "quarterly": False})
Returns annual income statements for Apple Inc.
"""
try:
ticker = yf.Ticker(params.symbol)
# Fetch the appropriate financial statement
if params.statement_type == FinancialStatement.INCOME_STATEMENT:
financials = ticker.quarterly_financials if params.quarterly else ticker.financials
statement_name = "Income Statement"
elif params.statement_type == FinancialStatement.BALANCE_SHEET:
financials = ticker.quarterly_balance_sheet if params.quarterly else ticker.balance_sheet
statement_name = "Balance Sheet"
else: # CASH_FLOW
financials = ticker.quarterly_cashflow if params.quarterly else ticker.cashflow
statement_name = "Cash Flow Statement"
if financials is None or financials.empty:
return json.dumps({
"error": f"No {statement_name.lower()} data available for '{params.symbol}'",
"suggestion": "This data may not be available for all securities. Try a different statement type or symbol."
}, indent=2)
# Convert DataFrame to dict
# Transpose so dates are keys and line items are nested
financials_dict = financials.to_dict()
# Format dates and convert to list of periods
periods = []
for date_col in financials.columns:
period_data = {
"date": date_col.strftime("%Y-%m-%d") if hasattr(date_col, 'strftime') else str(date_col),
"items": {}
}
for item_name, value in financials[date_col].items():
# Convert numpy types to Python types
if value is not None and not (isinstance(value, float) and str(value) == 'nan'):
period_data["items"][item_name] = float(value)
else:
period_data["items"][item_name] = None
periods.append(period_data)
summary = {
"symbol": params.symbol,
"statement_type": params.statement_type.value,
"period_type": "quarterly" if params.quarterly else "annual",
"number_of_periods": len(periods)
}
if params.response_format == ResponseFormat.JSON:
response = {
"summary": summary,
"periods": periods
}
content = json.dumps(response, indent=2)
return truncate_if_needed(content)
else:
# Markdown format
md = f"""# {statement_name}: {params.symbol}
## Summary
- **Statement Type:** {statement_name}
- **Period Type:** {'Quarterly' if params.quarterly else 'Annual'}
- **Number of Periods:** {summary['number_of_periods']}
## Financial Data
"""
# Show latest period in detail
if periods:
latest = periods[0]
md += f"\n### Most Recent Period: {latest['date']}\n\n"
# Sort items by absolute value for relevance
sorted_items = sorted(
latest['items'].items(),
key=lambda x: abs(x[1]) if x[1] is not None else 0,
reverse=True
)
# Show top items
for item_name, value in sorted_items[:20]: # Top 20 items
if value is not None:
md += f"- **{item_name}:** {format_currency(value)}\n"
if len(sorted_items) > 20:
md += f"\n*Showing top 20 of {len(sorted_items)} line items. Use JSON format for complete data.*\n"
# Show previous periods summary
if len(periods) > 1:
md += f"\n### Additional Periods Available\n"
for period in periods[1:]:
md += f"- {period['date']}\n"
md += f"\n*Use JSON format to access all periods and line items.*\n"
return truncate_if_needed(md)
except Exception as e:
error_response = {
"error": f"Failed to fetch financial statements for '{params.symbol}'",
"details": str(e),
"suggestion": "Verify the ticker symbol and try again. Some symbols may not have financial data available."
}
return json.dumps(error_response, indent=2)
@mcp.tool(
name="search_ticker",
annotations={
"title": "Search for Ticker Symbols",
"readOnlyHint": True,
"destructiveHint": False,
"idempotentHint": True,
"openWorldHint": True
}
)
async def search_ticker(params: TickerSearchInput) -> str:
"""Search for stock ticker symbols by company name.
Searches Yahoo Finance for ticker symbols matching a company name query.
Useful when you know the company name but need to find its ticker symbol.
Args:
params (TickerSearchInput): Validated input parameters containing:
- query (str): Company name or partial name to search
- response_format (ResponseFormat): Output format ('markdown' or 'json')
Returns:
str: Search results including:
- Ticker symbol
- Company name
- Exchange
- Security type
Example:
>>> search_ticker({"query": "Tesla"})
Returns ticker symbols for companies matching "Tesla"
Note:
This tool uses a basic search approach. For best results, use specific
company names. If no results are found, try alternative names or verify
the company is publicly traded.
"""
try:
# Use yfinance's Ticker to attempt a direct lookup
# Note: yfinance doesn't have a built-in search function, so we'll try a direct lookup
# and provide guidance if it doesn't work
# Try common ticker patterns
test_symbols = [
params.query.upper(), # Direct symbol
params.query.upper().replace(" ", ""), # Remove spaces
params.query.upper()[:4], # First 4 letters
]
results = []
for symbol in test_symbols:
try:
ticker = yf.Ticker(symbol)
info = ticker.info
if info and 'symbol' in info and info.get('longName'):
# Check if the company name matches the query
company_name = info.get('longName', '')
if params.query.lower() in company_name.lower() or company_name.lower() in params.query.lower():
results.append({
"symbol": info.get('symbol', symbol),
"name": company_name,
"exchange": info.get('exchange'),
"type": info.get('quoteType')
})
except:
continue
# Remove duplicates
seen = set()
unique_results = []
for r in results:
if r['symbol'] not in seen:
seen.add(r['symbol'])
unique_results.append(r)
if not unique_results:
return json.dumps({
"error": f"No ticker symbols found for '{params.query}'",
"suggestion": "Try searching with different keywords, check the spelling, or try searching online for the ticker symbol directly. Note: This tool has limited search capabilities and works best with exact or near-exact company names."
}, indent=2)
if params.response_format == ResponseFormat.JSON:
return json.dumps({"results": unique_results}, indent=2)
else:
md = f"""# Ticker Search Results: "{params.query}"
Found {len(unique_results)} result(s):
"""
for result in unique_results:
md += f"""## {result['symbol']}
- **Company:** {result['name']}
- **Exchange:** {result['exchange'] or 'N/A'}
- **Type:** {result['type'] or 'N/A'}
"""
return md
except Exception as e:
error_response = {
"error": f"Search failed for query '{params.query}'",
"details": str(e),
"suggestion": "Try a different search term or check the company name spelling."
}
return json.dumps(error_response, indent=2)
# ============================================================================
# Server Entry Point
# ============================================================================
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Stock Data MCP Server")
parser.add_argument(
"--transport",
default="stdio",
choices=["stdio", "sse", "streamable-http"],
help="Transport mode (stdio for local, sse/streamable-http for Railway)"
)
parser.add_argument(
"--host",
default=os.getenv("HOST", "0.0.0.0"),
help="Host to bind to (default: 0.0.0.0)"
)
parser.add_argument(
"--port",
type=int,
default=int(os.getenv("PORT", "8000")),
help="Port to bind to (default: PORT env var or 8000)"
)
args = parser.parse_args()
# Configure settings for HTTP-based transports
if args.transport in {"sse", "streamable-http"}:
mcp.settings.host = args.host
mcp.settings.port = args.port
print(f"Starting Stock MCP server on {args.host}:{args.port} with {args.transport} transport")
# Run the server
mcp.run(transport=args.transport)