# Directory Structure
```
├── .dockerignore
├── .gitattributes
├── .github
│ ├── dependabot.yaml
│ └── workflows
│ └── ci.yaml
├── .gitignore
├── .pre-commit-config.yaml
├── .python-version
├── Dockerfile
├── LICENSE
├── pyproject.toml
├── README.md
├── src
│ └── mcp_yahoo_finance
│ ├── __init__.py
│ ├── __main__.py
│ ├── py.typed
│ ├── server.py
│ └── utils.py
├── tests
│ └── test_server.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.python-version:
--------------------------------------------------------------------------------
```
3.10
```
--------------------------------------------------------------------------------
/.gitattributes:
--------------------------------------------------------------------------------
```
Dockerfile linguist-generated=true
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info
# Virtual environments
.venv
.env
Makefile
```
--------------------------------------------------------------------------------
/.pre-commit-config.yaml:
--------------------------------------------------------------------------------
```yaml
# See https://pre-commit.com for more information
# See https://pre-commit.com/hooks.html for more hooks
repos:
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v3.2.0
hooks:
- id: trailing-whitespace
- id: end-of-file-fixer
- id: check-yaml
- id: check-added-large-files
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.11.5
hooks:
- id: ruff
args: [ --fix, --select, I ]
- id: ruff-format
```
--------------------------------------------------------------------------------
/.dockerignore:
--------------------------------------------------------------------------------
```
# Git
.git
.gitignore
.gitattributes
# CI
.codeclimate.yml
.travis.yml
.taskcluster.yml
.pre-commit-config.yaml
# Docker
docker-compose.yml
Dockerfile
.docker
.dockerignore
# Byte-compiled / optimized / DLL files
**/__pycache__/
**/*.py[cod]
# C extensions
*.so
# Distribution / packaging
.Python
env/
build/
develop-eggs/
dist/
downloads/
eggs/
lib/
lib64/
parts/
sdist/
var/
*.egg-info/
.installed.cfg
*.egg
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.coverage
.cache
nosetests.xml
coverage.xml
.pytest_cache/
tests/
# Linting
.ruff_cache/
# Translations
*.mo
*.pot
# Django stuff:
*.log
# Sphinx documentation
docs/_build/
# PyBuilder
target/
# Virtual environment
.env
.venv/
venv/
# PyCharm
.idea
# Python mode for VIM
.ropeproject
**/.ropeproject
# Vim swap files
**/*.swp
# VS Code
.vscode/
# Other
Makefile
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# MCP Yahoo Finance
A [Model Context Protocol](https://modelcontextprotocol.io) (MCP) server for Yahoo Finance interaction. This server provides tools to get pricing, company information and more.
> Please note that `mcp-yahoo-finance` is currently in early development. The functionality and available tools are subject to change and expansion as I continue to develop and improve the server.
## Installation
You don't need to manually install `mcp-yahoo-finance` if you use [`uv`](https://docs.astral.sh/uv/). We'll use [`uvx`](https://docs.astral.sh/uv/guides/tools/) to directly run `mcp-yahoo-finance`.
I would recommend using this method if you simply want to use the MCP server.
### Using pip
Using `pip`.
```sh
pip install mcp-yahoo-finance
```
### Using Git
You can also install the package after cloning the repository to your machine.
```sh
git clone [email protected]:maxscheijen/mcp-yahoo-finance.git
cd mcp-yahoo-finance
uv sync
```
## Configuration
### Claude Desktop
Add this to your `claude_desktop_config.json`:
```json
{
"mcpServers": {
"yahoo-finance": {
"command": "uvx",
"args": ["mcp-yahoo-finance"]
}
}
}
```
You can also use docker:
```json
{
"mcpServers": {
"yahoo-finance": {
"command": "docker",
"args": ["run", "-i", "--rm", "IMAGE"]
}
}
}
```
### VSCode
Add this to your `.vscode/mcp.json`:
```json
{
"servers": {
"yahoo-finance": {
"command": "uvx",
"args": ["mcp-yahoo-finance"]
}
}
}
```
## Examples of Questions
1. "What is the stock price of Apple?"
2. "What is the difference in stock price between Apple and Google?"
3. "How much did the stock price of Apple change between 2024-01-01 and 2025-01-01?"
4. "What are the available options expiration dates for AAPL?"
5. "Show me the options chain for AAPL expiring on 2024-01-19"
6. "What are the call and put options for Tesla?"
## Build
Docker:
```sh
docker build -t [IMAGE] .
```
## Test with MCP Inspector
```sh
npx @modelcontextprotocol/inspector uv run mcp-yahoo-finance
```
```
--------------------------------------------------------------------------------
/src/mcp_yahoo_finance/__main__.py:
--------------------------------------------------------------------------------
```python
from mcp_yahoo_finance import main
main()
```
--------------------------------------------------------------------------------
/src/mcp_yahoo_finance/__init__.py:
--------------------------------------------------------------------------------
```python
from mcp_yahoo_finance import server
def main():
import asyncio
asyncio.run(server.serve())
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/.github/dependabot.yaml:
--------------------------------------------------------------------------------
```yaml
version: 2
updates:
- package-ecosystem: "pip"
directory: "/"
schedule:
interval: "weekly"
open-pull-requests-limit: 10
versioning-strategy: "increase"
```
--------------------------------------------------------------------------------
/.github/workflows/ci.yaml:
--------------------------------------------------------------------------------
```yaml
name: CI
on:
push:
branches: [ "main" ]
pull_request:
branches: [ "main" ]
jobs:
build:
runs-on: ubuntu-latest
strategy:
fail-fast: false
matrix:
python-version:
- "3.10"
- "3.11"
- "3.12"
- "3.13"
steps:
- uses: actions/checkout@v4
- name: Install uv and set the python version
uses: astral-sh/setup-uv@v6
with:
python-version: ${{ matrix.python-version }}
- name: Install Dependencies
run: |
uv sync --locked --all-extras
- name: Lint & Format with ruff
run: |
uv run ruff check .
uv run ruff format --check
```
--------------------------------------------------------------------------------
/Dockerfile:
--------------------------------------------------------------------------------
```dockerfile
# Use a Python image with uv pre-installed
FROM ghcr.io/astral-sh/uv:python3.13-bookworm-slim AS builder
# Install the project into `/app`
WORKDIR /app
# Enable bytecode compilation
ENV UV_COMPILE_BYTECODE=1
# Copy from the cache instead of linking since it's a mounted volume
ENV UV_LINK_MODE=copy
# Install the project's dependencies using the lockfile and settings
RUN --mount=type=cache,target=/root/.cache/uv \
--mount=type=bind,source=uv.lock,target=uv.lock \
--mount=type=bind,source=pyproject.toml,target=pyproject.toml \
uv sync --frozen --no-install-project --no-dev --no-editable
# Then, add the rest of the project source code and install it
# Installing separately from its dependencies allows optimal layer caching
ADD . /app
RUN --mount=type=cache,target=/root/.cache/uv \
uv sync --frozen --no-dev --no-editable
# Place executables in the environment at the front of the path
ENV PATH="/app/.venv/bin:$PATH"
# Set entry point
ENTRYPOINT ["mcp-yahoo-finance"]
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "mcp-yahoo-finance"
version = "0.1.3"
description = "A Model Context Protol (MCP) server providing tools to interact with Yahoo Finance for LLMs"
readme = {file = "README.md", content-type = "text/markdown"}
authors = [
{ name = "Max Scheijen", email = "[email protected]" }
]
maintainers = [
{name = "Max Scheijen", email = "[email protected]"}
]
requires-python = ">=3.10"
keywords = ["finance", "yahoo finance", "mcp", "llm", "automation"]
license = { text = "MIT" }
classifiers = [
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
]
dependencies = [
"mcp>=1.6.0",
"yfinance>=0.2.55",
]
[project.scripts]
mcp-yahoo-finance = "mcp_yahoo_finance:main"
[project.urls]
Homepage = "https://github.com/maxscheijen/mcp-yahoo-finance"
Documentation = "https://github.com/maxscheijen/mcp-yahoo-finance"
Repository = "https://github.com/maxscheijen/mcp-yahoo-finance.git"
Issues = "https://github.com/maxscheijen/mcp-yahoo-finance/issues"
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[dependency-groups]
dev = [
"pre-commit>=4.2.0",
"pytest-asyncio>=0.26.0",
"pytest>=8.3.5",
"ruff>=0.11.5",
]
```
--------------------------------------------------------------------------------
/src/mcp_yahoo_finance/utils.py:
--------------------------------------------------------------------------------
```python
import inspect
from typing import Any
from mcp.types import Tool
def parse_docstring(docstring: str) -> dict[str, str]:
"""Parses a Google-style docstring to extract parameter descriptions."""
descriptions = {}
if not docstring:
return descriptions
lines = docstring.split("\n")
current_param = None
for line in lines:
line = line.strip()
if line.startswith("Args:"):
continue
elif line and "(" in line and ")" in line and ":" in line:
param = line.split("(")[0].strip()
desc = line.split("):")[1].strip()
descriptions[param] = desc
current_param = param
elif current_param and line:
descriptions[current_param] += " " + line.strip()
return descriptions
def generate_tool(func: Any) -> Tool:
"""Generates a tool schema from a Python function."""
signature = inspect.signature(func)
docstring = inspect.getdoc(func) or ""
param_descriptions = parse_docstring(docstring)
schema = {
"name": func.__name__,
"description": docstring.split("Args:")[0].strip(),
"inputSchema": {
"type": "object",
"properties": {},
},
}
for param_name, param in signature.parameters.items():
param_type = (
"number"
if param.annotation is float
else "string"
if param.annotation is str
else "string"
)
schema["inputSchema"]["properties"][param_name] = {
"type": param_type,
"description": param_descriptions.get(param_name, ""),
}
if "required" not in schema["inputSchema"]:
schema["inputSchema"]["required"] = [param_name]
else:
if "=" not in str(param):
schema["inputSchema"]["required"].append(param_name)
return Tool(**schema)
```
--------------------------------------------------------------------------------
/tests/test_server.py:
--------------------------------------------------------------------------------
```python
import asyncio
import json
import pytest
from mcp import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client
from mcp.types import TextContent, Tool
@pytest.fixture
def server_params():
return StdioServerParameters(command="mcp-yahoo-finance")
@pytest.fixture
def client_tools() -> list[Tool]:
server_params = StdioServerParameters(command="mcp-yahoo-finance")
async def _get_tools():
async with (
stdio_client(server_params) as (read, write),
ClientSession(read, write) as session,
):
await session.initialize()
tool_list_result = await session.list_tools()
return tool_list_result.tools
return asyncio.run(_get_tools())
@pytest.mark.asyncio
@pytest.mark.parametrize(
"tool_name",
[
"get_current_stock_price",
"get_stock_price_by_date",
"get_stock_price_date_range",
"get_historical_stock_prices",
"get_dividends",
"get_income_statement",
"get_cashflow",
"get_earning_dates",
"get_news",
"get_recommendations",
"get_option_expiration_dates",
"get_option_chain",
],
)
async def test_list_tools(client_tools: list[Tool], tool_name) -> None:
tool_names = [tool.name for tool in client_tools]
assert tool_name in tool_names
@pytest.mark.asyncio
@pytest.mark.parametrize(
"symbol, date, expected_price",
[
("AAPL", "2025-01-01", 243.5822),
("GOOG", "2025-02-01", 202.4094),
("META", "2025-02-01", 696.8401),
],
)
async def test_get_stock_price_by_date(server_params, symbol, date, expected_price):
async with (
stdio_client(server_params) as (read, write),
ClientSession(read, write) as session,
):
await session.initialize()
tool_result = await session.call_tool(
"get_stock_price_by_date", {"symbol": symbol, "date": date}
)
assert len(tool_result.content) == 1
assert isinstance(tool_result.content[0], TextContent)
data = json.loads(tool_result.content[0].text)
assert isinstance(data, float)
assert data == expected_price
```
--------------------------------------------------------------------------------
/src/mcp_yahoo_finance/server.py:
--------------------------------------------------------------------------------
```python
import json
from typing import Any, Literal
import pandas as pd
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import TextContent, Tool
from requests import Session
from yfinance import Ticker
from mcp_yahoo_finance.utils import generate_tool
class YahooFinance:
def __init__(self, session: Session | None = None, verify: bool = True) -> None:
self.session = session
if self.session:
self.session.verify = verify
def get_current_stock_price(self, symbol: str) -> str:
"""Get the current stock price based on stock symbol.
Args:
symbol (str): Stock symbol in Yahoo Finance format.
"""
stock = Ticker(ticker=symbol, session=self.session).info
current_price = stock.get(
"regularMarketPrice", stock.get("currentPrice", "N/A")
)
return (
f"{current_price:.4f}"
if current_price
else f"Couldn't fetch {symbol} current price"
)
def get_stock_price_by_date(self, symbol: str, date: str) -> str:
"""Get the stock price for a given stock symbol on a specific date.
Args:
symbol (str): Stock symbol in Yahoo Finance format.
date (str): The date in YYYY-MM-DD format.
"""
stock = Ticker(ticker=symbol, session=self.session)
price = stock.history(start=date, period="1d")
return f"{price.iloc[0]['Close']:.4f}"
def get_stock_price_date_range(
self, symbol: str, start_date: str, end_date: str
) -> str:
"""Get the stock prices for a given date range for a given stock symbol.
Args:
symbol (str): Stock symbol in Yahoo Finance format.
start_date (str): The start date in YYYY-MM-DD format.
end_date (str): The end date in YYYY-MM-DD format.
"""
stock = Ticker(ticker=symbol, session=self.session)
prices = stock.history(start=start_date, end=end_date)
prices.index = prices.index.astype(str)
return f"{prices['Close'].to_json(orient='index')}"
def get_historical_stock_prices(
self,
symbol: str,
period: Literal[
"1d", "5d", "1mo", "3mo", "6mo", "1y", "2y", "5y", "10y", "ytd", "max"
] = "1mo",
interval: Literal["1d", "5d", "1wk", "1mo", "3mo"] = "1d",
) -> str:
"""Get historical stock prices for a given stock symbol.
Args:
symbol (str): Stock symbol in Yahoo Finance format.
period (str): The period for historical data. Defaults to "1mo".
Valid periods: "1d", "5d", "1mo", "3mo", "6mo", "1y", "2y", "5y", "10y", "ytd", "max"
interval (str): The interval beween data points. Defaults to "1d".
Valid intervals: "1d", "5d", "1wk", "1mo", "3mo"
"""
stock = Ticker(ticker=symbol, session=self.session)
prices = stock.history(period=period, interval=interval)
if hasattr(prices.index, "date"):
prices.index = prices.index.date.astype(str) # type: ignore
return f"{prices['Close'].to_json(orient='index')}"
def get_dividends(self, symbol: str) -> str:
"""Get dividends for a given stock symbol.
Args:
symbol (str): Stock symbol in Yahoo Finance format.
"""
stock = Ticker(ticker=symbol, session=self.session)
dividends = stock.dividends
if hasattr(dividends.index, "date"):
dividends.index = dividends.index.date.astype(str) # type: ignore
return f"{dividends.to_json(orient='index')}"
def get_income_statement(
self, symbol: str, freq: Literal["yearly", "quarterly", "trainling"] = "yearly"
) -> str:
"""Get income statement for a given stock symbol.
Args:
symbol (str): Stock symbol in Yahoo Finance format.
freq (str): At what frequency to get cashflow statements. Defaults to "yearly".
Valid freqencies: "yearly", "quarterly", "trainling"
"""
stock = Ticker(ticker=symbol, session=self.session)
income_statement = stock.get_income_stmt(freq=freq, pretty=True)
if isinstance(income_statement, pd.DataFrame):
income_statement.columns = [
str(col.date()) for col in income_statement.columns
]
return f"{income_statement.to_json()}"
return f"{income_statement}"
def get_cashflow(
self, symbol: str, freq: Literal["yearly", "quarterly", "trainling"] = "yearly"
):
"""Get cashflow for a given stock symbol.
Args:
symbol (str): Stock symbol in Yahoo Finance format.
freq (str): At what frequency to get cashflow statements. Defaults to "yearly".
Valid freqencies: "yearly", "quarterly", "trainling"
"""
stock = Ticker(ticker=symbol, session=self.session)
cashflow = stock.get_cashflow(freq=freq, pretty=True)
if isinstance(cashflow, pd.DataFrame):
cashflow.columns = [str(col.date()) for col in cashflow.columns]
return f"{cashflow.to_json(indent=2)}"
return f"{cashflow}"
def get_earning_dates(self, symbol: str, limit: int = 12) -> str:
"""Get earning dates.
Args:
symbol (str): Stock symbol in Yahoo Finance format.
limit (int): max amount of upcoming and recent earnings dates to return. Default value 12 should return next 4 quarters and last 8 quarters. Increase if more history is needed.
"""
stock = Ticker(ticker=symbol, session=self.session)
earning_dates = stock.get_earnings_dates(limit=limit)
if isinstance(earning_dates, pd.DataFrame):
earning_dates.index = earning_dates.index.date.astype(str) # type: ignore
return f"{earning_dates.to_json(indent=2)}"
return f"{earning_dates}"
def get_news(self, symbol: str) -> str:
"""Get news for a given stock symbol.
Args:
symbol (str): Stock symbol in Yahoo Finance format.
"""
stock = Ticker(ticker=symbol, session=self.session)
return json.dumps(stock.news, indent=2)
def get_recommendations(self, symbol: str) -> str:
"""Get analyst recommendations for a given symbol.
Args:
symbol (str): Stock symbol in Yahoo Finance format.
"""
stock = Ticker(ticker=symbol, session=self.session)
recommendations = stock.get_recommendations()
print(recommendations)
if isinstance(recommendations, pd.DataFrame):
return f"{recommendations.to_json(orient='records', indent=2)}"
return f"{recommendations}"
def get_option_expiration_dates(self, symbol: str) -> str:
"""Get available options expiration dates for a given stock symbol.
Args:
symbol (str): Stock symbol in Yahoo Finance format.
"""
stock = Ticker(ticker=symbol, session=self.session)
expiration_dates = stock.options
return json.dumps(list(expiration_dates), indent=2)
def get_option_chain(self, symbol: str, expiration_date: str) -> str:
"""Get options chain for a specific expiration date.
Args:
symbol (str): Stock symbol in Yahoo Finance format.
expiration_date (str): Options expiration date in YYYY-MM-DD format.
"""
stock = Ticker(ticker=symbol, session=self.session)
option_chain = stock.option_chain(expiration_date)
result = {"calls": None, "puts": None, "underlying": option_chain.underlying}
if option_chain.calls is not None:
# Convert dates to strings for JSON serialization
calls_df = option_chain.calls.copy()
if "lastTradeDate" in calls_df.columns:
calls_df["lastTradeDate"] = calls_df["lastTradeDate"].astype(str)
result["calls"] = calls_df.to_dict(orient="records")
if option_chain.puts is not None:
# Convert dates to strings for JSON serialization
puts_df = option_chain.puts.copy()
if "lastTradeDate" in puts_df.columns:
puts_df["lastTradeDate"] = puts_df["lastTradeDate"].astype(str)
result["puts"] = puts_df.to_dict(orient="records")
return json.dumps(result, indent=2)
async def serve() -> None:
server = Server("mcp-yahoo-finance")
yf = YahooFinance()
@server.list_tools()
async def list_tools() -> list[Tool]:
return [
generate_tool(yf.get_current_stock_price),
generate_tool(yf.get_stock_price_by_date),
generate_tool(yf.get_stock_price_date_range),
generate_tool(yf.get_historical_stock_prices),
generate_tool(yf.get_dividends),
generate_tool(yf.get_income_statement),
generate_tool(yf.get_cashflow),
generate_tool(yf.get_earning_dates),
generate_tool(yf.get_news),
generate_tool(yf.get_recommendations),
generate_tool(yf.get_option_expiration_dates),
generate_tool(yf.get_option_chain),
]
@server.call_tool()
async def call_tool(name: str, args: dict[str, Any]) -> list[TextContent]:
match name:
case "get_current_stock_price":
price = yf.get_current_stock_price(**args)
return [TextContent(type="text", text=price)]
case "get_stock_price_by_date":
price = yf.get_stock_price_by_date(**args)
return [TextContent(type="text", text=price)]
case "get_stock_price_date_range":
price = yf.get_stock_price_date_range(**args)
return [TextContent(type="text", text=price)]
case "get_historical_stock_prices":
price = yf.get_historical_stock_prices(**args)
return [TextContent(type="text", text=price)]
case "get_dividends":
price = yf.get_dividends(**args)
return [TextContent(type="text", text=price)]
case "get_income_statement":
price = yf.get_income_statement(**args)
return [TextContent(type="text", text=price)]
case "get_cashflow":
price = yf.get_cashflow(**args)
return [TextContent(type="text", text=price)]
case "get_earning_dates":
price = yf.get_earning_dates(**args)
return [TextContent(type="text", text=price)]
case "get_news":
price = yf.get_news(**args)
return [TextContent(type="text", text=price)]
case "get_recommendations":
recommendations = yf.get_recommendations(**args)
return [TextContent(type="text", text=recommendations)]
case "get_option_expiration_dates":
dates = yf.get_option_expiration_dates(**args)
return [TextContent(type="text", text=dates)]
case "get_option_chain":
chain = yf.get_option_chain(**args)
return [TextContent(type="text", text=chain)]
case _:
raise ValueError(f"Unknown tool: {name}")
options = server.create_initialization_options()
async with stdio_server() as (read_stream, write_stream):
await server.run(read_stream, write_stream, options, raise_exceptions=True)
```