# Directory Structure
```
├── .env.example
├── .gitignore
├── .python-version
├── LICENSE
├── pyproject.toml
├── README.md
├── server.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.python-version:
--------------------------------------------------------------------------------
```
3.11
```
--------------------------------------------------------------------------------
/.env.example:
--------------------------------------------------------------------------------
```
# Get your Financial Datasets API key from https://financialdatasets.ai/
FINANCIAL_DATASETS_API_KEY=your-financial-datasets-api-key
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# 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/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# UV
# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
#uv.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
#poetry.lock
# pdm
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
#pdm.lock
# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
# in version control.
# https://pdm.fming.dev/latest/usage/project/#working-with-version-control
.pdm.toml
.pdm-python
.pdm-build/
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/
# PyPI configuration file
.pypirc
.DS_Store
# Ignore vscode
.vscode/
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# Financial Datasets MCP Server
## Introduction
This is a Model Context Protocol (MCP) server that provides access to stock market data from [Financial Datasets](https://www.financialdatasets.ai/).
It allows Claude and other AI assistants to retrieve income statements, balance sheets, cash flow statements, stock prices, and market news directly through the MCP interface.
## Available Tools
This MCP server provides the following tools:
- **get_income_statements**: Get income statements for a company.
- **get_balance_sheets**: Get balance sheets for a company.
- **get_cash_flow_statements**: Get cash flow statements for a company.
- **get_current_stock_price**: Get the current / latest price of a company.
- **get_historical_stock_prices**: Gets historical stock prices for a company.
- **get_company_news**: Get news for a company.
- **get_available_crypto_tickers**: Gets all available crypto tickers.
- **get_crypto_prices**: Gets historical prices for a crypto currency.
- **get_historical_crypto_prices**: Gets historical prices for a crypto currency.
- **get_current_crypto_price**: Get the current / latest price of a crypto currency.
## Setup
### Prerequisites
- Python 3.10 or higher
- [uv](https://github.com/astral-sh/uv) package manager
### Installation
1. Clone this repository:
```bash
git clone https://github.com/financial-datasets/mcp-server
cd mcp-server
```
2. If you don't have uv installed, install it:
```bash
# macOS/Linux
curl -LsSf https://astral.sh/uv/install.sh | sh
# Windows
curl -LsSf https://astral.sh/uv/install.ps1 | powershell
```
3. Install dependencies:
```bash
# Create virtual env and activate it
uv venv
source .venv/bin/activate # On Windows: .venv\Scripts\activate
# Install dependencies
uv add "mcp[cli]" httpx # On Windows: uv add mcp[cli] httpx
```
4. Set up environment variables:
```bash
# Create .env file for your API keys
cp .env.example .env
# Set API key in .env
FINANCIAL_DATASETS_API_KEY=your-financial-datasets-api-key
```
5. Run the server:
```bash
uv run server.py
```
## Connecting to Claude Desktop
1. Install [Claude Desktop](https://claude.ai/desktop) if you haven't already
2. Create or edit the Claude Desktop configuration file:
```bash
# macOS
mkdir -p ~/Library/Application\ Support/Claude/
nano ~/Library/Application\ Support/Claude/claude_desktop_config.json
```
3. Add the following configuration:
```json
{
"mcpServers": {
"financial-datasets": {
"command": "/path/to/uv",
"args": [
"--directory",
"/absolute/path/to/financial-datasets-mcp",
"run",
"server.py"
]
}
}
}
```
Replace `/path/to/uv` with the result of `which uv` and `/absolute/path/to/financial-datasets-mcp` with the absolute path to this project.
4. Restart Claude Desktop
5. You should now see the financial tools available in Claude Desktop's tools menu (hammer icon)
6. Try asking Claude questions like:
- "What are Apple's recent income statements?"
- "Show me the current price of Tesla stock"
- "Get historical prices for MSFT from 2024-01-01 to 2024-12-31"
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "mcp-server"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
"httpx>=0.28.1",
"mcp[cli]>=1.3.0",
"python-dotenv>=1.0.0",
]
```
--------------------------------------------------------------------------------
/server.py:
--------------------------------------------------------------------------------
```python
import json
import os
import httpx
import logging
import sys
from mcp.server.fastmcp import FastMCP
from dotenv import load_dotenv
# Configure logging to write to stderr
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
stream=sys.stderr,
)
logger = logging.getLogger("financial-datasets-mcp")
# Initialize FastMCP server
mcp = FastMCP("financial-datasets")
# Constants
FINANCIAL_DATASETS_API_BASE = "https://api.financialdatasets.ai"
# Helper function to make API requests
async def make_request(url: str) -> dict[str, any] | None:
"""Make a request to the Financial Datasets API with proper error handling."""
# Load environment variables from .env file
load_dotenv()
headers = {}
if api_key := os.environ.get("FINANCIAL_DATASETS_API_KEY"):
headers["X-API-KEY"] = api_key
async with httpx.AsyncClient() as client:
try:
response = await client.get(url, headers=headers, timeout=30.0)
response.raise_for_status()
return response.json()
except Exception as e:
return {"Error": str(e)}
@mcp.tool()
async def get_income_statements(
ticker: str,
period: str = "annual",
limit: int = 4,
) -> str:
"""Get income statements for a company.
Args:
ticker: Ticker symbol of the company (e.g. AAPL, GOOGL)
period: Period of the income statement (e.g. annual, quarterly, ttm)
limit: Number of income statements to return (default: 4)
"""
# Fetch data from the API
url = f"{FINANCIAL_DATASETS_API_BASE}/financials/income-statements/?ticker={ticker}&period={period}&limit={limit}"
data = await make_request(url)
# Check if data is found
if not data:
return "Unable to fetch income statements or no income statements found."
# Extract the income statements
income_statements = data.get("income_statements", [])
# Check if income statements are found
if not income_statements:
return "Unable to fetch income statements or no income statements found."
# Stringify the income statements
return json.dumps(income_statements, indent=2)
@mcp.tool()
async def get_balance_sheets(
ticker: str,
period: str = "annual",
limit: int = 4,
) -> str:
"""Get balance sheets for a company.
Args:
ticker: Ticker symbol of the company (e.g. AAPL, GOOGL)
period: Period of the balance sheet (e.g. annual, quarterly, ttm)
limit: Number of balance sheets to return (default: 4)
"""
# Fetch data from the API
url = f"{FINANCIAL_DATASETS_API_BASE}/financials/balance-sheets/?ticker={ticker}&period={period}&limit={limit}"
data = await make_request(url)
# Check if data is found
if not data:
return "Unable to fetch balance sheets or no balance sheets found."
# Extract the balance sheets
balance_sheets = data.get("balance_sheets", [])
# Check if balance sheets are found
if not balance_sheets:
return "Unable to fetch balance sheets or no balance sheets found."
# Stringify the balance sheets
return json.dumps(balance_sheets, indent=2)
@mcp.tool()
async def get_cash_flow_statements(
ticker: str,
period: str = "annual",
limit: int = 4,
) -> str:
"""Get cash flow statements for a company.
Args:
ticker: Ticker symbol of the company (e.g. AAPL, GOOGL)
period: Period of the cash flow statement (e.g. annual, quarterly, ttm)
limit: Number of cash flow statements to return (default: 4)
"""
# Fetch data from the API
url = f"{FINANCIAL_DATASETS_API_BASE}/financials/cash-flow-statements/?ticker={ticker}&period={period}&limit={limit}"
data = await make_request(url)
# Check if data is found
if not data:
return "Unable to fetch cash flow statements or no cash flow statements found."
# Extract the cash flow statements
cash_flow_statements = data.get("cash_flow_statements", [])
# Check if cash flow statements are found
if not cash_flow_statements:
return "Unable to fetch cash flow statements or no cash flow statements found."
# Stringify the cash flow statements
return json.dumps(cash_flow_statements, indent=2)
@mcp.tool()
async def get_current_stock_price(ticker: str) -> str:
"""Get the current / latest price of a company.
Args:
ticker: Ticker symbol of the company (e.g. AAPL, GOOGL)
"""
# Fetch data from the API
url = f"{FINANCIAL_DATASETS_API_BASE}/prices/snapshot/?ticker={ticker}"
data = await make_request(url)
# Check if data is found
if not data:
return "Unable to fetch current price or no current price found."
# Extract the current price
snapshot = data.get("snapshot", {})
# Check if current price is found
if not snapshot:
return "Unable to fetch current price or no current price found."
# Stringify the current price
return json.dumps(snapshot, indent=2)
@mcp.tool()
async def get_historical_stock_prices(
ticker: str,
start_date: str,
end_date: str,
interval: str = "day",
interval_multiplier: int = 1,
) -> str:
"""Gets historical stock prices for a company.
Args:
ticker: Ticker symbol of the company (e.g. AAPL, GOOGL)
start_date: Start date of the price data (e.g. 2020-01-01)
end_date: End date of the price data (e.g. 2020-12-31)
interval: Interval of the price data (e.g. minute, hour, day, week, month)
interval_multiplier: Multiplier of the interval (e.g. 1, 2, 3)
"""
# Fetch data from the API
url = f"{FINANCIAL_DATASETS_API_BASE}/prices/?ticker={ticker}&interval={interval}&interval_multiplier={interval_multiplier}&start_date={start_date}&end_date={end_date}"
data = await make_request(url)
# Check if data is found
if not data:
return "Unable to fetch prices or no prices found."
# Extract the prices
prices = data.get("prices", [])
# Check if prices are found
if not prices:
return "Unable to fetch prices or no prices found."
# Stringify the prices
return json.dumps(prices, indent=2)
@mcp.tool()
async def get_company_news(ticker: str) -> str:
"""Get news for a company.
Args:
ticker: Ticker symbol of the company (e.g. AAPL, GOOGL)
"""
# Fetch data from the API
url = f"{FINANCIAL_DATASETS_API_BASE}/news/?ticker={ticker}"
data = await make_request(url)
# Check if data is found
if not data:
return "Unable to fetch news or no news found."
# Extract the news
news = data.get("news", [])
# Check if news are found
if not news:
return "Unable to fetch news or no news found."
return json.dumps(news, indent=2)
@mcp.tool()
async def get_available_crypto_tickers() -> str:
"""
Gets all available crypto tickers.
"""
# Fetch data from the API
url = f"{FINANCIAL_DATASETS_API_BASE}/crypto/prices/tickers"
data = await make_request(url)
# Check if data is found
if not data:
return "Unable to fetch available crypto tickers or no available crypto tickers found."
# Extract the available crypto tickers
tickers = data.get("tickers", [])
# Stringify the available crypto tickers
return json.dumps(tickers, indent=2)
@mcp.tool()
async def get_crypto_prices(
ticker: str,
start_date: str,
end_date: str,
interval: str = "day",
interval_multiplier: int = 1,
) -> str:
"""
Gets historical prices for a crypto currency.
"""
# Fetch data from the API
url = f"{FINANCIAL_DATASETS_API_BASE}/crypto/prices/?ticker={ticker}&interval={interval}&interval_multiplier={interval_multiplier}&start_date={start_date}&end_date={end_date}"
data = await make_request(url)
# Check if data is found
if not data:
return "Unable to fetch prices or no prices found."
# Extract the prices
prices = data.get("prices", [])
# Check if prices are found
if not prices:
return "Unable to fetch prices or no prices found."
# Stringify the prices
return json.dumps(prices, indent=2)
@mcp.tool()
async def get_historical_crypto_prices(
ticker: str,
start_date: str,
end_date: str,
interval: str = "day",
interval_multiplier: int = 1,
) -> str:
"""Gets historical prices for a crypto currency.
Args:
ticker: Ticker symbol of the crypto currency (e.g. BTC-USD). The list of available crypto tickers can be retrieved via the get_available_crypto_tickers tool.
start_date: Start date of the price data (e.g. 2020-01-01)
end_date: End date of the price data (e.g. 2020-12-31)
interval: Interval of the price data (e.g. minute, hour, day, week, month)
interval_multiplier: Multiplier of the interval (e.g. 1, 2, 3)
"""
# Fetch data from the API
url = f"{FINANCIAL_DATASETS_API_BASE}/crypto/prices/?ticker={ticker}&interval={interval}&interval_multiplier={interval_multiplier}&start_date={start_date}&end_date={end_date}"
data = await make_request(url)
# Check if data is found
if not data:
return "Unable to fetch prices or no prices found."
# Extract the prices
prices = data.get("prices", [])
# Check if prices are found
if not prices:
return "Unable to fetch prices or no prices found."
# Stringify the prices
return json.dumps(prices, indent=2)
@mcp.tool()
async def get_current_crypto_price(ticker: str) -> str:
"""Get the current / latest price of a crypto currency.
Args:
ticker: Ticker symbol of the crypto currency (e.g. BTC-USD). The list of available crypto tickers can be retrieved via the get_available_crypto_tickers tool.
"""
# Fetch data from the API
url = f"{FINANCIAL_DATASETS_API_BASE}/crypto/prices/snapshot/?ticker={ticker}"
data = await make_request(url)
# Check if data is found
if not data:
return "Unable to fetch current price or no current price found."
# Extract the current price
snapshot = data.get("snapshot", {})
# Check if current price is found
if not snapshot:
return "Unable to fetch current price or no current price found."
# Stringify the current price
return json.dumps(snapshot, indent=2)
@mcp.tool()
async def get_sec_filings(
ticker: str,
limit: int = 10,
filing_type: str | None = None,
) -> str:
"""Get all SEC filings for a company.
Args:
ticker: Ticker symbol of the company (e.g. AAPL, GOOGL)
limit: Number of SEC filings to return (default: 10)
filing_type: Type of SEC filing (e.g. 10-K, 10-Q, 8-K)
"""
# Fetch data from the API
url = f"{FINANCIAL_DATASETS_API_BASE}/filings/?ticker={ticker}&limit={limit}"
if filing_type:
url += f"&filing_type={filing_type}"
# Call the API
data = await make_request(url)
# Extract the SEC filings
filings = data.get("filings", [])
# Check if SEC filings are found
if not filings:
return f"Unable to fetch SEC filings or no SEC filings found."
# Stringify the SEC filings
return json.dumps(filings, indent=2)
if __name__ == "__main__":
# Log server startup
logger.info("Starting Financial Datasets MCP Server...")
# Initialize and run the server
mcp.run(transport="stdio")
# This line won't be reached during normal operation
logger.info("Server stopped")
```