main.py•4.16 kB
import yfinance as yf
from mcp.server.fastmcp import FastMCP, Context
from datetime import datetime, timedelta
from tabulate import tabulate
# Initialize the MCP server
mcp = FastMCP("CryptoStocksMCP", dependencies=["yfinance", "tabulate"])
# List of crypto-related stocks (example: companies related to cryptocurrency)
CRYPTO_STOCKS = {
"BMNR": "BitMine Immersion Technologies, Inc.",
"CRCL": "Circle Internet Group Inc.",
"SBET": "SharpLink Gaming Inc.",
"SRM": "Tron Inc.",
"DFDV": "DeFi Development Corp.",
"MSTR": "MicroStrategy Incorporated",
"COIN": "Coinbase Global Inc.",
"MARA": "Marathon Digital Holdings",
"RIOT": "Riot Platforms Inc.",
"HIVE": "HIVE Digital Technologies",
"CORZ": "Core Scientific Inc.",
"IREN": "Iris Energy Limited",
"CLSK": "CleanSpark Inc.",
"HUT": "Hut 8 Corp",
"CIFR": "Cipher Mining Inc.",
"BITF": "Bitfarms Ltd"
}
# Tool: List all crypto-related stocks
@mcp.tool()
def get_stock_list() -> str:
"""Return a list of available crypto-related stocks in a table format.
Returns:
str: An ASCII table string containing the ticker and name of crypto-related stocks.
"""
table_data = [
[ticker, name]
for ticker, name in CRYPTO_STOCKS.items()
]
return tabulate(
table_data,
headers=["Ticker", "Name"],
tablefmt="grid",
stralign="left"
)
# Tool: Get real-time price for a specific stock
@mcp.tool()
def get_stock_price(ticker: str) -> str:
"""Get real-time price for a specific stock in a table format.
Args:
ticker (str): The stock ticker symbol (e.g., 'COIN' for Coinbase).
Returns:
str: An ASCII table string containing the ticker, company name, current price, and timestamp.
Returns an error message if the ticker is invalid.
"""
if ticker not in CRYPTO_STOCKS:
return f"Error: Unknown ticker {ticker}"
stock = yf.Ticker(ticker)
price = stock.info.get("regularMarketPrice", "N/A")
table_data = [[ticker, CRYPTO_STOCKS[ticker], price, datetime.now().isoformat()]]
return tabulate(
table_data,
headers=["Ticker", "Name", "Price", "Timestamp"],
tablefmt="grid",
stralign="left",
floatfmt=".2f"
)
# Tool: Fetch historical stock prices for a given ticker
@mcp.tool()
def get_historical_prices(ticker: str, days: int = 30) -> str:
"""Fetch historical stock prices for the given ticker over the specified number of days in a table format.
Args:
ticker (str): The stock ticker symbol (e.g., 'COIN' for Coinbase).
days (int, optional): Number of days for historical data (default: 30).
Returns:
str: An ASCII table string containing the date, closing price, and volume for the specified stock.
Returns an error message if the ticker is invalid.
"""
if ticker not in CRYPTO_STOCKS:
return f"Error: Unknown ticker {ticker}"
stock = yf.Ticker(ticker)
end_date = datetime.now()
start_date = end_date - timedelta(days=days)
hist = stock.history(start=start_date, end=end_date)
# Format historical data as a table
table_data = [
[index.strftime("%Y-%m-%d"), row["Close"], row["Volume"]]
for index, row in hist.iterrows()
]
return tabulate(
table_data,
headers=["Date", "Close", "Volume"],
tablefmt="grid",
stralign="left",
floatfmt=".2f"
)
# Prompt: Generate a query for stock information
@mcp.prompt()
def stock_query(ticker: str) -> str:
"""Generate a prompt for querying stock information.
Args:
ticker (str): The stock ticker symbol (e.g., 'COIN' for Coinbase).
Returns:
str: A prompt string for querying stock information, or an error message if the ticker is invalid.
"""
if ticker not in CRYPTO_STOCKS:
return f"Error: Unknown ticker {ticker}. Available tickers: {', '.join(CRYPTO_STOCKS.keys())}"
return f"Get the latest price and historical data for {CRYPTO_STOCKS[ticker]} ({ticker})."
# Main program: Run the server
if __name__ == "__main__":
mcp.run()