# Directory Structure
```
├── .dockerignore
├── .gitignore
├── .python-version
├── CLAUDE.md
├── Dockerfile
├── LICENSE
├── pyproject.toml
├── README.md
├── smithery.yaml
├── src
│ └── alpha_vantage_mcp
│ ├── __init__.py
│ ├── server.py
│ └── tools.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.python-version:
--------------------------------------------------------------------------------
```
3.12
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info
# Virtual environments
.venv
```
--------------------------------------------------------------------------------
/.dockerignore:
--------------------------------------------------------------------------------
```
# Version control
.git/
.gitignore
.github/
# Environment
.env
.venv/
venv/
env/
.python-version
# IDE and editor files
.vscode/
.idea/
*.swp
*.swo
*~
.DS_Store
# Project configuration
smithery.yaml
.dockerignore
# Documentation
README.md
LICENSE
docs/
*.md
# Testing
tests/
**/__pycache__/
**/*.pyc
**/*.pyo
**/*.pyd
**/.pytest_cache/
**/.coverage
htmlcov/
# Build artifacts
*.egg-info/
dist/
build/
# Temporary files
*.log
*.tmp
.cache/
temp/
# Keep only what's needed
# (Don't ignore these files)
!pyproject.toml
!uv.lock
!src/
!entrypoint.sh
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# Alpha Vantage MCP Server
[](https://smithery.ai/server/@berlinbra/alpha-vantage-mcp)
A Model Context Protocol (MCP) server that provides real-time access to financial market data through the free [Alpha Vantage API](https://www.alphavantage.co/documentation/). This server implements a standardized interface for retrieving stock quotes and company information.
<a href="https://glama.ai/mcp/servers/0wues5td08"><img width="380" height="200" src="https://glama.ai/mcp/servers/0wues5td08/badge" alt="AlphaVantage-MCP MCP server" /></a>
# Features
- Real-time stock quotes with price, volume, and change data
- Detailed company information including sector, industry, and market cap
- Real-time cryptocurrency exchange rates with bid/ask prices
- Daily, weekly, and monthly cryptocurrency time series data
- Real-time options chain data with Greeks and implied volatility
- Historical options chain data with advanced filtering and sorting
- Comprehensive ETF profile data with holdings, sector allocation, and key metrics
- Upcoming earnings calendar with customizable time horizons
- Historical earnings data with annual and quarterly reports
- Built-in error handling and rate limit management
## Installation
### Using Claude Desktop
#### Installing via Docker
- Clone the repository and build a local image to be utilized by your Claude desktop client
```sh
cd alpha-vantage-mcp
docker build -t mcp/alpha-vantage .
```
- Change your `claude_desktop_config.json` to match the following, replacing `REPLACE_API_KEY` with your actual key:
> `claude_desktop_config.json` path
>
> - On MacOS: `~/Library/Application\ Support/Claude/claude_desktop_config.json`
> - On Windows: `%APPDATA%/Claude/claude_desktop_config.json`
```json
{
"mcpServers": {
"alphavantage": {
"command": "docker",
"args": [
"run",
"-i",
"-e",
"ALPHA_VANTAGE_API_KEY",
"mcp/alpha-vantage"
],
"env": {
"ALPHA_VANTAGE_API_KEY": "REPLACE_API_KEY"
}
}
}
}
```
#### Installing via Smithery
To install Alpha Vantage MCP Server for Claude Desktop automatically via [Smithery](https://smithery.ai/server/@berlinbra/alpha-vantage-mcp):
```bash
npx -y @smithery/cli install @berlinbra/alpha-vantage-mcp --client claude
```
<summary> <h3> Development/Unpublished Servers Configuration <h3> </summary>
<details>
```json
{
"mcpServers": {
"alpha-vantage-mcp": {
"args": [
"--directory",
"/Users/{INSERT_USER}/YOUR/PATH/TO/alpha-vantage-mcp",
"run",
"alpha-vantage-mcp"
],
"command": "uv",
"env": {
"ALPHA_VANTAGE_API_KEY": "<insert api key>"
}
}
}
}
```
</details>
#### Install packages
```
uv install -e .
```
#### Running
After connecting Claude client with the MCP tool via json file and installing the packages, Claude should see the server's mcp tools:
You can run the sever yourself via:
In alpha-vantage-mcp repo:
```
uv run src/alpha_vantage_mcp/server.py
```
with inspector
```
* npx @modelcontextprotocol/inspector uv --directory /Users/{INSERT_USER}/YOUR/PATH/TO/alpha-vantage-mcp run src/alpha_vantage_mcp/server.py `
```
## Available Tools
The server implements twelve tools:
- `get-stock-quote`: Get the latest stock quote for a specific company
- `get-company-info`: Get stock-related information for a specific company
- `get-crypto-exchange-rate`: Get current cryptocurrency exchange rates
- `get-time-series`: Get historical daily price data for a stock
- `get-realtime-options`: Get real-time options chain data with Greeks and implied volatility
- `get-historical-options`: Get historical options chain data with advanced filtering and sorting capabilities
- `get-etf-profile`: Get comprehensive ETF profile information including holdings and sector allocation
- `get-crypto-daily`: Get daily time series data for a cryptocurrency
- `get-crypto-weekly`: Get weekly time series data for a cryptocurrency
- `get-crypto-monthly`: Get monthly time series data for a cryptocurrency
- `get-earnings-calendar`: Get upcoming earnings calendar data for companies
- `get-historical-earnings`: Get historical earnings data for a specific company
### get-stock-quote
**Input Schema:**
```json
{
"symbol": {
"type": "string",
"description": "Stock symbol (e.g., AAPL, MSFT)"
}
}
```
**Example Response:**
```
Stock quote for AAPL:
Price: $198.50
Change: $2.50 (+1.25%)
Volume: 58942301
High: $199.62
Low: $197.20
```
### get-company-info
Retrieves detailed company information for a given symbol.
**Input Schema:**
```json
{
"symbol": {
"type": "string",
"description": "Stock symbol (e.g., AAPL, MSFT)"
}
}
```
**Example Response:**
```
Company information for AAPL:
Name: Apple Inc
Sector: Technology
Industry: Consumer Electronics
Market Cap: $3000000000000
Description: Apple Inc. designs, manufactures, and markets smartphones...
Exchange: NASDAQ
Currency: USD
```
### get-crypto-exchange-rate
Retrieves real-time cryptocurrency exchange rates with additional market data.
**Input Schema:**
```json
{
"crypto_symbol": {
"type": "string",
"description": "Cryptocurrency symbol (e.g., BTC, ETH)"
},
"market": {
"type": "string",
"description": "Market currency (e.g., USD, EUR)",
"default": "USD"
}
}
```
**Example Response:**
```
Cryptocurrency exchange rate for BTC/USD:
From: Bitcoin (BTC)
To: United States Dollar (USD)
Exchange Rate: 43521.45000
Last Updated: 2024-12-17 19:45:00 UTC
Bid Price: 43521.00000
Ask Price: 43522.00000
```
### get-time-series
Retrieves daily time series (OHLCV) data with optional date filtering.
**Input Schema:**
```json
{
"symbol": {
"type": "string",
"description": "Stock symbol (e.g., AAPL, MSFT)"
},
"outputsize": {
"type": "string",
"description": "compact (latest 100 data points) or full (up to 20 years of data). When start_date or end_date is specified, defaults to 'full'",
"default": "compact"
},
"start_date": {
"type": "string",
"description": "Optional: Start date in YYYY-MM-DD format for filtering results",
"pattern": "^20[0-9]{2}-(?:0[1-9]|1[0-2])-(?:0[1-9]|[12][0-9]|3[01])$"
},
"end_date": {
"type": "string",
"description": "Optional: End date in YYYY-MM-DD format for filtering results",
"pattern": "^20[0-9]{2}-(?:0[1-9]|1[0-2])-(?:0[1-9]|[12][0-9]|3[01])$"
},
"limit": {
"type": "integer",
"description": "Optional: Number of data points to return when no date filtering is applied (default: 5)",
"default": 5,
"minimum": 1
}
}
```
**Example Response (Recent Data):**
```
Time Series Data for AAPL (Last Refreshed: 2024-12-17 16:00:00):
(Showing 5 most recent data points)
Date: 2024-12-16
Open: $195.09
High: $197.68
Low: $194.83
Close: $197.57
Volume: 55,751,011
---
Date: 2024-12-13
Open: $194.50
High: $196.25
Low: $193.80
Close: $195.12
Volume: 48,320,567
---
```
**Example Response (Date Range Filtering):**
```
Time Series Data for AAPL (Last Refreshed: 2024-12-17 16:00:00):
Date Range: 2024-12-01 to 2024-12-07 (5 data points)
Date: 2024-12-06
Open: $191.25
High: $193.80
Low: $190.55
Close: $192.90
Volume: 52,145,890
---
Date: 2024-12-05
Open: $189.75
High: $192.40
Low: $188.90
Close: $191.30
Volume: 47,892,345
---
```
### get-realtime-options
Retrieves real-time options chain data for a stock with optional Greeks calculation and contract filtering.
**⚠️ PREMIUM SUBSCRIPTION REQUIRED**: This endpoint requires Alpha Vantage Premium with either the 600 requests/minute or 1200 requests/minute plan. The standard 75 requests/minute plan and free accounts will receive placeholder/demo data instead of real market data. For most use cases, consider using `get-historical-options` which works with all API key tiers.
**Input Schema:**
```json
{
"symbol": {
"type": "string",
"description": "Stock symbol (e.g., AAPL, MSFT)"
},
"require_greeks": {
"type": "boolean",
"description": "Optional: Enable Greeks and implied volatility calculation (default: false)",
"default": false
},
"contract": {
"type": "string",
"description": "Optional: Specific options contract ID to retrieve"
},
"datatype": {
"type": "string",
"description": "Optional: Response format (json or csv, default: json)",
"enum": ["json", "csv"],
"default": "json"
}
}
```
**Example Response:**
```
Realtime Options Data for AAPL
Last Updated: 2025-01-21 16:00:00
=== Expiration: 2025-01-24 ===
Strike: $220.0 (CALL)
Last: $5.25
Bid: $5.10
Ask: $5.30
Volume: 1250
Open Interest: 8420
IV: 0.28
Delta: 0.65
Gamma: 0.02
Theta: -0.15
Vega: 0.45
Rho: 0.12
---
Strike: $220.0 (PUT)
Last: $1.85
Bid: $1.80
Ask: $1.90
Volume: 820
Open Interest: 5240
IV: 0.25
Delta: -0.35
Gamma: 0.02
Theta: -0.12
Vega: 0.42
Rho: -0.08
---
```
**Note**: The above example shows real market data which is only available with Alpha Vantage Premium 600+ requests/minute plans. Users with free accounts or 75 requests/minute plans will see placeholder data (symbols like "XXYYZZ", dates like "2099-99-99") and should use `get-historical-options` instead.
### get-historical-options
Retrieves historical options chain data with advanced filtering and sorting capabilities to find specific contracts.
**Input Schema:**
```json
{
"symbol": {
"type": "string",
"description": "Stock symbol (e.g., AAPL, MSFT)"
},
"date": {
"type": "string",
"description": "Optional: Trading date in YYYY-MM-DD format (defaults to previous trading day, must be after 2008-01-01)",
"pattern": "^20[0-9]{2}-(?:0[1-9]|1[0-2])-(?:0[1-9]|[12][0-9]|3[01])$"
},
"expiry_date": {
"type": "string",
"description": "Optional: Filter by expiration date in YYYY-MM-DD format",
"pattern": "^20[0-9]{2}-(?:0[1-9]|1[0-2])-(?:0[1-9]|[12][0-9]|3[01])$"
},
"min_strike": {
"type": "number",
"description": "Optional: Minimum strike price filter (e.g., 100.00)",
"minimum": 0
},
"max_strike": {
"type": "number",
"description": "Optional: Maximum strike price filter (e.g., 200.00)",
"minimum": 0
},
"contract_id": {
"type": "string",
"description": "Optional: Filter by specific contract ID (e.g., MSTR260116C00000500)"
},
"contract_type": {
"type": "string",
"description": "Optional: Filter by contract type (call or put)",
"enum": ["call", "put", "C", "P"]
},
"limit": {
"type": "integer",
"description": "Optional: Number of contracts to return after filtering (default: 10, use -1 for all contracts)",
"default": 10,
"minimum": -1
},
"sort_by": {
"type": "string",
"description": "Optional: Field to sort by",
"enum": ["strike", "expiration", "volume", "open_interest", "implied_volatility", "delta", "gamma", "theta", "vega", "rho", "last", "bid", "ask"],
"default": "strike"
},
"sort_order": {
"type": "string",
"description": "Optional: Sort order",
"enum": ["asc", "desc"],
"default": "asc"
}
}
```
**Example Response (Basic):**
```
Historical Options Data for AAPL (2024-02-20):
Status: success
Found 156 contracts, sorted by: strike (asc)
Contract Details:
Contract ID: AAPL240315C00190000
Expiration: 2024-03-15
Strike: $190.00
Type: call
Last: $8.45
Bid: $8.40
Ask: $8.50
Volume: 1245
Open Interest: 4567
Implied Volatility: 0.25
Greeks:
Delta: 0.65
Gamma: 0.04
Theta: -0.15
Vega: 0.30
Rho: 0.25
---
```
**Example Response (Filtered):**
```
Historical Options Data for MSTR (2024-02-20):
Status: success
Filters: Expiry: 2026-01-16, Strike: min $400 - max $600, Type: call
Found 3 contracts, sorted by: strike (asc)
Contract Details:
Contract ID: MSTR260116C00000500
Expiration: 2026-01-16
Strike: $500.00
Type: call
Last: $125.30
Bid: $124.50
Ask: $126.10
Volume: 89
Open Interest: 1234
---
```
### get-etf-profile
Retrieves comprehensive ETF profile information including basic metrics, sector allocation, and top holdings.
**Input Schema:**
```json
{
"symbol": {
"type": "string",
"description": "ETF symbol (e.g., QQQ, SPY, VTI)"
}
}
```
**Example Response:**
```
ETF profile for QQQ:
ETF Profile
Basic Information:
Net Assets: $352,700,000,000
Net Expense Ratio: 0.200%
Portfolio Turnover: 8.0%
Dividend Yield: 0.50%
Inception Date: 1999-03-10
Leveraged: NO
Sector Allocation:
INFORMATION TECHNOLOGY: 51.9%
COMMUNICATION SERVICES: 15.4%
CONSUMER DISCRETIONARY: 12.2%
CONSUMER STAPLES: 4.8%
HEALTHCARE: 4.5%
INDUSTRIALS: 4.4%
UTILITIES: 1.4%
MATERIALS: 1.3%
ENERGY: 0.5%
FINANCIALS: 0.4%
Top Holdings:
1. NVDA - NVIDIA CORP: 9.80%
2. MSFT - MICROSOFT CORP: 8.85%
3. AAPL - APPLE INC: 7.35%
4. AMZN - AMAZON.COM INC: 5.65%
5. AVGO - BROADCOM INC: 5.14%
6. META - META PLATFORMS INC CLASS A: 3.63%
7. NFLX - NETFLIX INC: 3.10%
8. TSLA - TESLA INC: 2.66%
9. GOOGL - ALPHABET INC CLASS A: 2.49%
10. COST - COSTCO WHOLESALE CORP: 2.49%
... and 92 more holdings
Total Holdings: 102
```
### get-crypto-daily
Retrieves daily time series data for a cryptocurrency.
**Input Schema:**
```json
{
"symbol": {
"type": "string",
"description": "Cryptocurrency symbol (e.g., BTC, ETH)"
},
"market": {
"type": "string",
"description": "Market currency (e.g., USD, EUR)",
"default": "USD"
}
}
```
**Example Response:**
```
Daily cryptocurrency time series for SOL in USD:
Daily Time Series for Solana (SOL)
Market: United States Dollar (USD)
Last Refreshed: 2025-04-17 00:00:00 UTC
Date: 2025-04-17
Open: 131.31000000 USD
High: 131.67000000 USD
Low: 130.74000000 USD
Close: 131.15000000 USD
Volume: 39652.22195178
---
Date: 2025-04-16
Open: 126.10000000 USD
High: 133.91000000 USD
Low: 123.46000000 USD
Close: 131.32000000 USD
Volume: 1764240.04195810
---
```
### get-crypto-weekly
Retrieves weekly time series data for a cryptocurrency.
**Input Schema:**
```json
{
"symbol": {
"type": "string",
"description": "Cryptocurrency symbol (e.g., BTC, ETH)"
},
"market": {
"type": "string",
"description": "Market currency (e.g., USD, EUR)",
"default": "USD"
}
}
```
**Example Response:**
```
Weekly cryptocurrency time series for SOL in USD:
Weekly Time Series for Solana (SOL)
Market: United States Dollar (USD)
Last Refreshed: 2025-04-17 00:00:00 UTC
Date: 2025-04-17
Open: 128.32000000 USD
High: 136.00000000 USD
Low: 123.46000000 USD
Close: 131.15000000 USD
Volume: 4823091.05667581
---
Date: 2025-04-13
Open: 105.81000000 USD
High: 134.11000000 USD
Low: 95.16000000 USD
Close: 128.32000000 USD
Volume: 18015328.38860037
---
```
### get-crypto-monthly
Retrieves monthly time series data for a cryptocurrency.
**Input Schema:**
```json
{
"symbol": {
"type": "string",
"description": "Cryptocurrency symbol (e.g., BTC, ETH)"
},
"market": {
"type": "string",
"description": "Market currency (e.g., USD, EUR)",
"default": "USD"
}
}
```
**Example Response:**
```
Monthly cryptocurrency time series for SOL in USD:
Monthly Time Series for Solana (SOL)
Market: United States Dollar (USD)
Last Refreshed: 2025-04-17 00:00:00 UTC
Date: 2025-04-17
Open: 124.51000000 USD
High: 136.18000000 USD
Low: 95.16000000 USD
Close: 131.15000000 USD
Volume: 34268628.85976021
---
Date: 2025-03-31
Open: 148.09000000 USD
High: 180.00000000 USD
Low: 112.00000000 USD
Close: 124.54000000 USD
Volume: 42360395.75443056
---
```
### get-earnings-calendar
Retrieves upcoming earnings calendar data for companies with customizable time horizons and sorting capabilities.
**Input Schema:**
```json
{
"symbol": {
"type": "string",
"description": "Optional: Stock symbol to filter earnings for a specific company (e.g., AAPL, MSFT, IBM)"
},
"horizon": {
"type": "string",
"description": "Optional: Time horizon for earnings data (3month, 6month, or 12month)",
"enum": ["3month", "6month", "12month"],
"default": "12month"
},
"limit": {
"type": "integer",
"description": "Optional: Number of earnings entries to return (default: 100)",
"default": 100,
"minimum": 1
},
"sort_by": {
"type": "string",
"description": "Optional: Field to sort by",
"enum": ["reportDate", "symbol", "name", "fiscalDateEnding", "estimate"],
"default": "reportDate"
},
"sort_order": {
"type": "string",
"description": "Optional: Sort order",
"enum": ["asc", "desc"],
"default": "desc"
}
}
```
**Example Response (Default - Latest First):**
```
Earnings calendar (12month):
Upcoming Earnings Calendar (Sorted by reportDate desc):
Company: NVDA - NVIDIA Corp
Report Date: 2025-08-15
Fiscal Date End: 2025-07-31
Estimate: $4.25 USD
---
Company: AAPL - Apple Inc
Report Date: 2025-07-30
Fiscal Date End: 2025-06-30
Estimate: $1.85 USD
---
Company: MSTR - MicroStrategy Inc
Report Date: 2025-05-08
Fiscal Date End: 2025-03-31
Estimate: $1.30 USD
---
Company: MSTR - MicroStrategy Inc
Report Date: 2025-02-06
Fiscal Date End: 2024-12-31
Estimate: $1.25 USD
---
```
**Example Response (Sorted by Symbol):**
```
Earnings calendar (12month):
Upcoming Earnings Calendar (Sorted by symbol asc):
Company: AAPL - Apple Inc
Report Date: 2025-07-30
Fiscal Date End: 2025-06-30
Estimate: $1.85 USD
---
Company: GOOGL - Alphabet Inc
Report Date: 2025-04-25
Fiscal Date End: 2025-03-31
Estimate: $2.15 USD
---
Company: MSTR - MicroStrategy Inc
Report Date: 2025-02-06
Fiscal Date End: 2024-12-31
Estimate: $1.25 USD
---
```
### get-historical-earnings
Retrieves historical earnings data for a specific company, including both annual and quarterly reports.
**Input Schema:**
```json
{
"symbol": {
"type": "string",
"description": "Stock symbol for the company (e.g., AAPL, MSFT, IBM)"
},
"limit_annual": {
"type": "integer",
"description": "Optional: Number of annual earnings to return (default: 5)",
"default": 5,
"minimum": 1
},
"limit_quarterly": {
"type": "integer",
"description": "Optional: Number of quarterly earnings to return (default: 8)",
"default": 8,
"minimum": 1
}
}
```
**Example Response:**
```
Historical Earnings for MSTR:
=== ANNUAL EARNINGS ===
Fiscal Year End: 2023-12-31
Reported EPS: $5.40
---
Fiscal Year End: 2022-12-31
Reported EPS: $-9.98
---
=== QUARTERLY EARNINGS ===
Fiscal Quarter End: 2024-09-30
Reported Date: 2024-10-30
Reported EPS: $1.10
Estimated EPS: $0.98
Surprise: +$0.12 (+12.24%)
Report Time: post-market
---
Fiscal Quarter End: 2024-06-30
Reported Date: 2024-08-01
Reported EPS: $1.05
Estimated EPS: $0.92
Surprise: +$0.13 (+14.13%)
Report Time: post-market
---
```
## Error Handling
The server includes comprehensive error handling for various scenarios:
- Rate limit exceeded
- Invalid API key
- Network connectivity issues
- Timeout handling
- Malformed responses
Error messages are returned in a clear, human-readable format.
## Prerequisites
- Python 3.12 or higher
- httpx
- mcp
## Contributors
- [berlinbra](https://github.com/berlinbra)
- [zzulanas](https://github.com/zzulanas)
## Contributing
Contributions are welcome! Please feel free to submit a Pull Request.
## License
This MCP server is licensed under the MIT License.
This means you are free to use, modify, and distribute the software, subject to the terms and conditions of the MIT License. For more details, please see the LICENSE file in the project repository.
```
--------------------------------------------------------------------------------
/CLAUDE.md:
--------------------------------------------------------------------------------
```markdown
# Alpha Vantage MCP Server - Claude.md
## Overview
The Alpha Vantage MCP Server is a Model Context Protocol (MCP) implementation that provides real-time access to financial market data through the Alpha Vantage API. This server enables Claude and other AI assistants to retrieve stock quotes, company information, and cryptocurrency data.
## Architecture
### Core Components
- **server.py** - Main MCP server implementation that handles tool registration and execution
- **tools.py** - Utility functions for API requests and data formatting
- **smithery.yaml** - Smithery MCP configuration for automated deployment
### Key Features
1. **Stock Market Data**
- Real-time stock quotes with price, volume, and change data
- Detailed company information (sector, industry, market cap)
- Historical daily time series data (OHLCV)
2. **Cryptocurrency Support**
- Real-time cryptocurrency exchange rates with bid/ask prices
- Daily, weekly, and monthly cryptocurrency time series data
3. **Advanced Options Data**
- Historical options chain data with Greeks calculations
- Advanced filtering and sorting capabilities by strike, volume, IV, etc.
4. **Error Handling & Rate Limiting**
- Built-in error handling for API failures
- Rate limit management and timeout handling
- Comprehensive error messages
- **CRITICAL**: Never use placeholder data in responses - always handle errors properly and transparently
## Available Tools
The server implements 12 MCP tools:
1. **get-stock-quote** - Current stock quote information
2. **get-company-info** - Detailed company overview data
3. **get-crypto-exchange-rate** - Cryptocurrency exchange rates
4. **get-time-series** - Historical daily stock price data
5. **get-realtime-options** - Real-time options chain data (premium required)
6. **get-historical-options** - Options chain data with sorting
7. **get-etf-profile** - Comprehensive ETF profile with holdings and sector allocation
8. **get-crypto-daily** - Daily cryptocurrency time series
9. **get-crypto-weekly** - Weekly cryptocurrency time series
10. **get-crypto-monthly** - Monthly cryptocurrency time series
11. **get-earnings-calendar** - Upcoming earnings calendar data
12. **get-historical-earnings** - Historical earnings data
## Smithery MCP Integration
### Configuration (smithery.yaml:3-18)
The repository uses Smithery for automated MCP deployment and configuration:
- **startCommand**: Defines STDIO-based communication
- **configSchema**: JSON Schema requiring `alphaVantageApiKey`
- **commandFunction**: JavaScript function that generates the CLI command with proper environment variables
### Key Smithery Features
- **Automated Installation**: One-command install via `npx @smithery/cli install @berlinbra/alpha-vantage-mcp --client claude`
- **Configuration Management**: Handles API key injection and environment setup
- **Claude Desktop Integration**: Direct integration with Claude desktop client
### Smithery Benefits
1. **Simplified Deployment**: No manual configuration of `claude_desktop_config.json`
2. **Environment Management**: Automatic API key handling
3. **Version Management**: Centralized distribution and updates
4. **Cross-Platform Support**: Works on macOS, Windows, and Linux
## Installation Methods
### 1. Via Smithery (Recommended)
```bash
npx -y @smithery/cli install @berlinbra/alpha-vantage-mcp --client claude
```
### 2. Via Docker
- Build local image and configure `claude_desktop_config.json`
- Requires manual environment variable setup
### 3. Development Setup
- Use `uv` package manager for local development
- Requires manual path configuration
## API Integration
### Alpha Vantage API (tools.py:12-70)
- **Base URL**: `https://www.alphavantage.co/query`
- **Authentication**: API key via `ALPHA_VANTAGE_API_KEY` environment variable
- **Request Handling**: Async HTTP client with 30-second timeout
- **Error Management**: Comprehensive error handling for rate limits, authentication, and network issues
### Supported API Functions
**Free Tier Compatible:**
- `GLOBAL_QUOTE` - Stock quotes
- `OVERVIEW` - Company information
- `CURRENCY_EXCHANGE_RATE` - Crypto exchange rates
- `TIME_SERIES_DAILY` - Historical stock data
- `HISTORICAL_OPTIONS` - Options chain data
- `ETF_PROFILE` - ETF profile data with holdings and sectors
- `EARNINGS_CALENDAR` - Upcoming earnings data
- `EARNINGS` - Historical earnings data
- `DIGITAL_CURRENCY_DAILY/WEEKLY/MONTHLY` - Crypto time series
**Premium Subscription Required:**
- `REALTIME_OPTIONS` - Real-time options chain data (returns demo data with free API keys)
## Development Commands
### Running the Server
```bash
# Direct execution
uv run src/alpha_vantage_mcp/server.py
# With MCP Inspector
npx @modelcontextprotocol/inspector uv --directory /path/to/alpha-vantage-mcp run src/alpha_vantage_mcp/server.py
```
### Package Management
```bash
uv install -e .
```
## Dependencies
- **Python 3.12+** - Required runtime
- **httpx** - Async HTTP client for API requests
- **mcp** - Model Context Protocol framework
## Contributors
- [berlinbra](https://github.com/berlinbra) - Primary maintainer
- [zzulanas](https://github.com/zzulanas) - Contributor
## Error Handling Guidelines
### Core Principles
1. **No Placeholder Data**: Under no circumstances should the server return placeholder or demo data as if it were real market data
2. **Transparent Error Reporting**: All API errors, rate limits, and access issues must be clearly communicated to users
3. **Proper Error Detection**: The server must detect and flag demo/placeholder responses from the API
4. **User Education**: Error messages should help users understand API limitations and subscription requirements
### Common Error Scenarios
- **Rate Limiting**: Alpha Vantage free tier has request limits
- **Premium Features**: Some endpoints (like realtime options) require paid subscriptions
- **API Key Issues**: Invalid or expired API keys
- **Demo Data**: Alpha Vantage may return placeholder data for demo purposes
- **Network Issues**: Connection timeouts and service unavailability
### Implementation Requirements
All formatting functions in `tools.py` must:
- Detect placeholder/demo data patterns (e.g., "XXYYZZ" symbols, "2099-99-99" dates)
- Return clear error messages instead of formatting fake data
- Provide guidance on resolving access issues
- Log error patterns for debugging
## License
MIT License - Free for use, modification, and distribution
```
--------------------------------------------------------------------------------
/src/alpha_vantage_mcp/__init__.py:
--------------------------------------------------------------------------------
```python
from . import server
import asyncio
def main():
"""Main entry point for the package."""
asyncio.run(server.main())
# Optionally expose other important items at package level
__all__ = ['main', 'server']
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "alpha-vantage-mcp"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
"httpx>=0.28.1",
"mcp>=1.1.2",
]
[build-system]
requires = [ "hatchling",]
build-backend = "hatchling.build"
[project.scripts]
alpha-vantage-mcp = "alpha_vantage_mcp:main"
[[project.authors]]
name = "berlinbra"
email = "[email protected]"
```
--------------------------------------------------------------------------------
/smithery.yaml:
--------------------------------------------------------------------------------
```yaml
# Smithery configuration file: https://smithery.ai/docs/config#smitheryyaml
startCommand:
type: stdio
configSchema:
# JSON Schema defining the configuration options for the MCP.
type: object
required:
- alphaVantageApiKey
properties:
alphaVantageApiKey:
type: string
description: Your Alpha Vantage API Key for accessing the API.
commandFunction:
# A function that produces the CLI command to start the MCP on stdio.
|-
(config) => ({ command: 'python', args: ['-m', 'src.alpha_vantage_mcp.server'], env: { ALPHA_VANTAGE_API_KEY: config.alphaVantageApiKey } })
```
--------------------------------------------------------------------------------
/Dockerfile:
--------------------------------------------------------------------------------
```dockerfile
# Generated by https://smithery.ai. See: https://smithery.ai/docs/config#dockerfile
# Start with a base Python image
FROM python:3.12-slim-bookworm
# Set the working directory in the container
WORKDIR /app
# Copy the pyproject.toml and uv.lock for dependencies
COPY pyproject.toml uv.lock /app/
# Install dependencies
RUN pip install uvicorn 'httpx>=0.28.1' 'mcp>=1.1.2'
# Copy the rest of the application code
COPY src/ /app/src/
# Set environment variables
ENV ALPHA_VANTAGE_API_KEY=${ALPHA_VANTAGE_API_KEY}
# Expose the port that the app runs on
EXPOSE 8000
# Run the application
CMD ["python", "-m", "src.alpha_vantage_mcp.server"]
```
--------------------------------------------------------------------------------
/src/alpha_vantage_mcp/server.py:
--------------------------------------------------------------------------------
```python
from typing import Any, List, Dict, Optional
import asyncio
import httpx
from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server
import mcp.server.stdio
import os
# Import functions from tools.py
from .tools import (
make_alpha_request,
format_quote,
format_company_info,
format_crypto_rate,
format_time_series,
format_historical_options,
format_crypto_time_series,
format_earnings_calendar,
format_historical_earnings,
format_realtime_options,
format_etf_profile,
ALPHA_VANTAGE_BASE,
API_KEY
)
if not API_KEY:
raise ValueError("Missing ALPHA_VANTAGE_API_KEY environment variable")
server = Server("alpha_vantage_finance")
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
"""
List available tools.
Each tool specifies its arguments using JSON Schema validation.
"""
return [
types.Tool(
name="get-stock-quote",
description="Get current stock quote information",
inputSchema={
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "Stock symbol (e.g., AAPL, MSFT)",
},
},
"required": ["symbol"],
},
),
types.Tool(
name="get-company-info",
description="Get detailed company information",
inputSchema={
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "Stock symbol (e.g., AAPL, MSFT)",
},
},
"required": ["symbol"],
},
),
types.Tool(
name="get-crypto-exchange-rate",
description="Get current cryptocurrency exchange rate",
inputSchema={
"type": "object",
"properties": {
"crypto_symbol": {
"type": "string",
"description": "Cryptocurrency symbol (e.g., BTC, ETH)",
},
"market": {
"type": "string",
"description": "Market currency (e.g., USD, EUR)",
"default": "USD"
}
},
"required": ["crypto_symbol"],
},
),
types.Tool(
name="get-time-series",
description="Get daily time series data for a stock with optional date filtering",
inputSchema={
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "Stock symbol (e.g., AAPL, MSFT)",
},
"outputsize": {
"type": "string",
"description": "compact (latest 100 data points) or full (up to 20 years of data). When start_date or end_date is specified, defaults to 'full'",
"enum": ["compact", "full"],
"default": "compact"
},
"start_date": {
"type": "string",
"description": "Optional: Start date in YYYY-MM-DD format for filtering results",
"pattern": "^20[0-9]{2}-(?:0[1-9]|1[0-2])-(?:0[1-9]|[12][0-9]|3[01])$"
},
"end_date": {
"type": "string",
"description": "Optional: End date in YYYY-MM-DD format for filtering results",
"pattern": "^20[0-9]{2}-(?:0[1-9]|1[0-2])-(?:0[1-9]|[12][0-9]|3[01])$"
},
"limit": {
"type": "integer",
"description": "Optional: Number of data points to return when no date filtering is applied (default: 5)",
"default": 5,
"minimum": 1
}
},
"required": ["symbol"],
},
),
types.Tool(
name="get-historical-options",
description="Get historical options chain data for a stock with advanced filtering and sorting capabilities",
inputSchema={
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "Stock symbol (e.g., AAPL, MSFT)",
},
"date": {
"type": "string",
"description": "Optional: Trading date in YYYY-MM-DD format (defaults to previous trading day, must be after 2008-01-01)",
"pattern": "^20[0-9]{2}-(?:0[1-9]|1[0-2])-(?:0[1-9]|[12][0-9]|3[01])$"
},
"expiry_date": {
"type": "string",
"description": "Optional: Filter by expiration date in YYYY-MM-DD format",
"pattern": "^20[0-9]{2}-(?:0[1-9]|1[0-2])-(?:0[1-9]|[12][0-9]|3[01])$"
},
"min_strike": {
"type": "number",
"description": "Optional: Minimum strike price filter (e.g., 100.00)",
"minimum": 0
},
"max_strike": {
"type": "number",
"description": "Optional: Maximum strike price filter (e.g., 200.00)",
"minimum": 0
},
"contract_id": {
"type": "string",
"description": "Optional: Filter by specific contract ID (e.g., MSTR260116C00000500)"
},
"contract_type": {
"type": "string",
"description": "Optional: Filter by contract type (call or put)",
"enum": ["call", "put", "C", "P"]
},
"limit": {
"type": "integer",
"description": "Optional: Number of contracts to return after filtering (default: 10, use -1 for all contracts)",
"default": 10,
"minimum": -1
},
"sort_by": {
"type": "string",
"description": "Optional: Field to sort by",
"enum": [
"strike",
"expiration",
"volume",
"open_interest",
"implied_volatility",
"delta",
"gamma",
"theta",
"vega",
"rho",
"last",
"bid",
"ask"
],
"default": "strike"
},
"sort_order": {
"type": "string",
"description": "Optional: Sort order",
"enum": ["asc", "desc"],
"default": "asc"
}
},
"required": ["symbol"],
},
),
types.Tool(
name="get-crypto-daily",
description="Get daily time series data for a cryptocurrency",
inputSchema={
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "Cryptocurrency symbol (e.g., BTC, ETH)",
},
"market": {
"type": "string",
"description": "Market currency (e.g., USD, EUR)",
"default": "USD"
}
},
"required": ["symbol"],
},
),
types.Tool(
name="get-crypto-weekly",
description="Get weekly time series data for a cryptocurrency",
inputSchema={
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "Cryptocurrency symbol (e.g., BTC, ETH)",
},
"market": {
"type": "string",
"description": "Market currency (e.g., USD, EUR)",
"default": "USD"
}
},
"required": ["symbol"],
},
),
types.Tool(
name="get-crypto-monthly",
description="Get monthly time series data for a cryptocurrency",
inputSchema={
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "Cryptocurrency symbol (e.g., BTC, ETH)",
},
"market": {
"type": "string",
"description": "Market currency (e.g., USD, EUR)",
"default": "USD"
}
},
"required": ["symbol"],
},
),
types.Tool(
name="get-earnings-calendar",
description="Get upcoming earnings calendar data for companies with sorting capabilities",
inputSchema={
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "Optional: Stock symbol to filter earnings for a specific company (e.g., AAPL, MSFT, IBM)"
},
"horizon": {
"type": "string",
"description": "Optional: Time horizon for earnings data (3month, 6month, or 12month)",
"enum": ["3month", "6month", "12month"],
"default": "12month"
},
"limit": {
"type": "integer",
"description": "Optional: Number of earnings entries to return (default: 100)",
"default": 100,
"minimum": 1
},
"sort_by": {
"type": "string",
"description": "Optional: Field to sort by",
"enum": ["reportDate", "symbol", "name", "fiscalDateEnding", "estimate"],
"default": "reportDate"
},
"sort_order": {
"type": "string",
"description": "Optional: Sort order",
"enum": ["asc", "desc"],
"default": "desc"
}
},
"required": [],
},
),
types.Tool(
name="get-historical-earnings",
description="Get historical earnings data for a specific company",
inputSchema={
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "Stock symbol for the company (e.g., AAPL, MSFT, IBM)"
},
"limit_annual": {
"type": "integer",
"description": "Optional: Number of annual earnings to return (default: 5)",
"default": 5,
"minimum": 1
},
"limit_quarterly": {
"type": "integer",
"description": "Optional: Number of quarterly earnings to return (default: 8)",
"default": 8,
"minimum": 1
}
},
"required": ["symbol"],
},
),
types.Tool(
name="get-realtime-options",
description="Get realtime options chain data for a stock with optional Greeks and filtering",
inputSchema={
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "Stock symbol (e.g., AAPL, MSFT)",
},
"require_greeks": {
"type": "boolean",
"description": "Optional: Enable Greeks and implied volatility calculation (default: false)",
"default": False
},
"contract": {
"type": "string",
"description": "Optional: Specific options contract ID to retrieve"
},
"datatype": {
"type": "string",
"description": "Optional: Response format (json or csv, default: json)",
"enum": ["json", "csv"],
"default": "json"
}
},
"required": ["symbol"],
},
),
types.Tool(
name="get-etf-profile",
description="Get comprehensive ETF profile information including holdings, sector allocation, and key metrics",
inputSchema={
"type": "object",
"properties": {
"symbol": {
"type": "string",
"description": "ETF symbol (e.g., QQQ, SPY, VTI)",
}
},
"required": ["symbol"],
},
)
]
@server.call_tool()
async def handle_call_tool(
name: str, arguments: dict | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
"""
Handle tool execution requests.
Tools can fetch financial data and notify clients of changes.
"""
if not arguments:
return [types.TextContent(type="text", text="Missing arguments for the request")]
if name == "get-stock-quote":
symbol = arguments.get("symbol")
if not symbol:
return [types.TextContent(type="text", text="Missing symbol parameter")]
symbol = symbol.upper()
async with httpx.AsyncClient() as client:
quote_data = await make_alpha_request(
client,
"GLOBAL_QUOTE",
symbol
)
if isinstance(quote_data, str):
return [types.TextContent(type="text", text=f"Error: {quote_data}")]
formatted_quote = format_quote(quote_data)
quote_text = f"Stock quote for {symbol}:\n\n{formatted_quote}"
return [types.TextContent(type="text", text=quote_text)]
elif name == "get-company-info":
symbol = arguments.get("symbol")
if not symbol:
return [types.TextContent(type="text", text="Missing symbol parameter")]
symbol = symbol.upper()
async with httpx.AsyncClient() as client:
company_data = await make_alpha_request(
client,
"OVERVIEW",
symbol
)
if isinstance(company_data, str):
return [types.TextContent(type="text", text=f"Error: {company_data}")]
formatted_info = format_company_info(company_data)
info_text = f"Company information for {symbol}:\n\n{formatted_info}"
return [types.TextContent(type="text", text=info_text)]
elif name == "get-crypto-exchange-rate":
crypto_symbol = arguments.get("crypto_symbol")
if not crypto_symbol:
return [types.TextContent(type="text", text="Missing crypto_symbol parameter")]
market = arguments.get("market", "USD")
crypto_symbol = crypto_symbol.upper()
market = market.upper()
async with httpx.AsyncClient() as client:
crypto_data = await make_alpha_request(
client,
"CURRENCY_EXCHANGE_RATE",
None,
{
"from_currency": crypto_symbol,
"to_currency": market
}
)
if isinstance(crypto_data, str):
return [types.TextContent(type="text", text=f"Error: {crypto_data}")]
formatted_rate = format_crypto_rate(crypto_data)
rate_text = f"Cryptocurrency exchange rate for {crypto_symbol}/{market}:\n\n{formatted_rate}"
return [types.TextContent(type="text", text=rate_text)]
elif name == "get-time-series":
symbol = arguments.get("symbol")
if not symbol:
return [types.TextContent(type="text", text="Missing symbol parameter")]
symbol = symbol.upper()
start_date = arguments.get("start_date")
end_date = arguments.get("end_date")
limit = arguments.get("limit", 5)
# Auto-select outputsize: use 'full' when date filtering is requested
outputsize = arguments.get("outputsize")
if not outputsize:
outputsize = "full" if (start_date or end_date) else "compact"
async with httpx.AsyncClient() as client:
time_series_data = await make_alpha_request(
client,
"TIME_SERIES_DAILY",
symbol,
{"outputsize": outputsize}
)
if isinstance(time_series_data, str):
return [types.TextContent(type="text", text=f"Error: {time_series_data}")]
formatted_series = format_time_series(time_series_data, start_date, end_date, limit)
series_text = f"Time series data for {symbol}:\n\n{formatted_series}"
return [types.TextContent(type="text", text=series_text)]
elif name == "get-historical-options":
symbol = arguments.get("symbol")
date = arguments.get("date")
expiry_date = arguments.get("expiry_date")
min_strike = arguments.get("min_strike")
max_strike = arguments.get("max_strike")
contract_id = arguments.get("contract_id")
contract_type = arguments.get("contract_type")
limit = arguments.get("limit", 10)
sort_by = arguments.get("sort_by", "strike")
sort_order = arguments.get("sort_order", "asc")
if not symbol:
return [types.TextContent(type="text", text="Missing symbol parameter")]
symbol = symbol.upper()
async with httpx.AsyncClient() as client:
params = {}
if date:
params["date"] = date
options_data = await make_alpha_request(
client,
"HISTORICAL_OPTIONS",
symbol,
params
)
if isinstance(options_data, str):
return [types.TextContent(type="text", text=f"Error: {options_data}")]
formatted_options = format_historical_options(
options_data,
limit,
sort_by,
sort_order,
expiry_date,
min_strike,
max_strike,
contract_id,
contract_type
)
options_text = f"Historical options data for {symbol}"
if date:
options_text += f" on {date}"
options_text += f":\n\n{formatted_options}"
return [types.TextContent(type="text", text=options_text)]
elif name == "get-crypto-daily":
symbol = arguments.get("symbol")
market = arguments.get("market", "USD")
if not symbol:
return [types.TextContent(type="text", text="Missing symbol parameter")]
symbol = symbol.upper()
market = market.upper()
async with httpx.AsyncClient() as client:
crypto_data = await make_alpha_request(
client,
"DIGITAL_CURRENCY_DAILY",
symbol,
{"market": market}
)
if isinstance(crypto_data, str):
return [types.TextContent(type="text", text=f"Error: {crypto_data}")]
formatted_data = format_crypto_time_series(crypto_data, "daily")
data_text = f"Daily cryptocurrency time series for {symbol} in {market}:\n\n{formatted_data}"
return [types.TextContent(type="text", text=data_text)]
elif name == "get-crypto-weekly":
symbol = arguments.get("symbol")
market = arguments.get("market", "USD")
if not symbol:
return [types.TextContent(type="text", text="Missing symbol parameter")]
symbol = symbol.upper()
market = market.upper()
async with httpx.AsyncClient() as client:
crypto_data = await make_alpha_request(
client,
"DIGITAL_CURRENCY_WEEKLY",
symbol,
{"market": market}
)
if isinstance(crypto_data, str):
return [types.TextContent(type="text", text=f"Error: {crypto_data}")]
formatted_data = format_crypto_time_series(crypto_data, "weekly")
data_text = f"Weekly cryptocurrency time series for {symbol} in {market}:\n\n{formatted_data}"
return [types.TextContent(type="text", text=data_text)]
elif name == "get-crypto-monthly":
symbol = arguments.get("symbol")
market = arguments.get("market", "USD")
if not symbol:
return [types.TextContent(type="text", text="Missing symbol parameter")]
symbol = symbol.upper()
market = market.upper()
async with httpx.AsyncClient() as client:
crypto_data = await make_alpha_request(
client,
"DIGITAL_CURRENCY_MONTHLY",
symbol,
{"market": market}
)
if isinstance(crypto_data, str):
return [types.TextContent(type="text", text=f"Error: {crypto_data}")]
formatted_data = format_crypto_time_series(crypto_data, "monthly")
data_text = f"Monthly cryptocurrency time series for {symbol} in {market}:\n\n{formatted_data}"
return [types.TextContent(type="text", text=data_text)]
elif name == "get-earnings-calendar":
symbol = arguments.get("symbol")
horizon = arguments.get("horizon", "12month")
limit = arguments.get("limit", 100)
sort_by = arguments.get("sort_by", "reportDate")
sort_order = arguments.get("sort_order", "desc")
async with httpx.AsyncClient() as client:
params = {"horizon": horizon}
if symbol:
params["symbol"] = symbol.upper()
earnings_data = await make_alpha_request(
client,
"EARNINGS_CALENDAR",
None,
params
)
if isinstance(earnings_data, str):
return [types.TextContent(type="text", text=f"Error: {earnings_data}")]
formatted_earnings = format_earnings_calendar(earnings_data, limit, sort_by, sort_order)
earnings_text = f"Earnings calendar"
if symbol:
earnings_text += f" for {symbol.upper()}"
if horizon:
earnings_text += f" ({horizon})"
earnings_text += f":\n\n{formatted_earnings}"
return [types.TextContent(type="text", text=earnings_text)]
elif name == "get-historical-earnings":
symbol = arguments.get("symbol")
limit_annual = arguments.get("limit_annual", 5)
limit_quarterly = arguments.get("limit_quarterly", 8)
if not symbol:
return [types.TextContent(type="text", text="Missing symbol parameter")]
symbol = symbol.upper()
async with httpx.AsyncClient() as client:
earnings_data = await make_alpha_request(
client,
"EARNINGS",
symbol
)
if isinstance(earnings_data, str):
return [types.TextContent(type="text", text=f"Error: {earnings_data}")]
formatted_earnings = format_historical_earnings(earnings_data, limit_annual, limit_quarterly)
earnings_text = f"Historical earnings for {symbol}:\n\n{formatted_earnings}"
return [types.TextContent(type="text", text=earnings_text)]
elif name == "get-realtime-options":
symbol = arguments.get("symbol")
require_greeks = arguments.get("require_greeks", False)
contract = arguments.get("contract")
datatype = arguments.get("datatype", "json")
if not symbol:
return [types.TextContent(type="text", text="Missing symbol parameter")]
symbol = symbol.upper()
async with httpx.AsyncClient() as client:
params = {}
if require_greeks:
params["require_greeks"] = "true"
if contract:
params["contract"] = contract
if datatype:
params["datatype"] = datatype
options_data = await make_alpha_request(
client,
"REALTIME_OPTIONS",
symbol,
params
)
if isinstance(options_data, str):
return [types.TextContent(type="text", text=f"Error: {options_data}")]
formatted_options = format_realtime_options(options_data)
options_text = f"Realtime options data for {symbol}:\n\n{formatted_options}"
return [types.TextContent(type="text", text=options_text)]
elif name == "get-etf-profile":
symbol = arguments.get("symbol")
if not symbol:
return [types.TextContent(type="text", text="Missing symbol parameter")]
symbol = symbol.upper()
async with httpx.AsyncClient() as client:
etf_data = await make_alpha_request(
client,
"ETF_PROFILE",
symbol
)
if isinstance(etf_data, str):
return [types.TextContent(type="text", text=f"Error: {etf_data}")]
formatted_etf = format_etf_profile(etf_data)
etf_text = f"ETF profile for {symbol}:\n\n{formatted_etf}"
return [types.TextContent(type="text", text=etf_text)]
else:
return [types.TextContent(type="text", text=f"Unknown tool: {name}")]
async def main():
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="alpha_vantage_finance",
server_version="0.1.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
# This is needed if you'd like to connect to a custom client
if __name__ == "__main__":
asyncio.run(main())
```
--------------------------------------------------------------------------------
/src/alpha_vantage_mcp/tools.py:
--------------------------------------------------------------------------------
```python
"""
Alpha Vantage MCP Tools Module
This module contains utility functions for making requests to the Alpha Vantage API
and formatting the responses.
"""
from typing import Any, Dict, Optional, List
import httpx
import os
import csv
import io
from datetime import datetime
ALPHA_VANTAGE_BASE = "https://www.alphavantage.co/query"
API_KEY = os.getenv('ALPHA_VANTAGE_API_KEY')
async def make_alpha_request(client: httpx.AsyncClient, function: str, symbol: Optional[str], additional_params: Optional[Dict[str, Any]] = None) -> Dict[str, Any] | str:
"""Make a request to the Alpha Vantage API with proper error handling.
Args:
client: An httpx AsyncClient instance
function: The Alpha Vantage API function to call
symbol: The stock/crypto symbol (can be None for some endpoints)
additional_params: Additional parameters to include in the request
Returns:
Either a dictionary containing the API response, or a string with an error message
"""
params = {
"function": function,
"apikey": API_KEY
}
if symbol:
params["symbol"] = symbol
if additional_params:
params.update(additional_params)
try:
response = await client.get(
ALPHA_VANTAGE_BASE,
params=params,
timeout=30.0
)
# Check for specific error responses
if response.status_code == 429:
return f"Rate limit exceeded. Error details: {response.text}"
elif response.status_code == 403:
return f"API key invalid or expired. Error details: {response.text}"
response.raise_for_status()
# Check if response is empty
if not response.text.strip():
return "Empty response received from Alpha Vantage API"
# Special handling for EARNINGS_CALENDAR which returns CSV by default
if function == "EARNINGS_CALENDAR":
try:
# Parse CSV response
csv_reader = csv.DictReader(io.StringIO(response.text))
earnings_list = list(csv_reader)
return earnings_list
except Exception as e:
return f"Error parsing CSV response: {str(e)}"
# For other functions, expect JSON
try:
data = response.json()
except ValueError as e:
return f"Invalid JSON response from Alpha Vantage API: {response.text[:200]}"
# Check for Alpha Vantage specific error messages
if "Error Message" in data:
return f"Alpha Vantage API error: {data['Error Message']}"
if "Note" in data and "API call frequency" in data["Note"]:
return f"Rate limit warning: {data['Note']}"
return data
except httpx.TimeoutException:
return "Request timed out after 30 seconds. The Alpha Vantage API may be experiencing delays."
except httpx.ConnectError:
return "Failed to connect to Alpha Vantage API. Please check your internet connection."
except httpx.HTTPStatusError as e:
return f"HTTP error occurred: {str(e)} - Response: {e.response.text}"
except Exception as e:
return f"Unexpected error occurred: {str(e)}"
def format_quote(quote_data: Dict[str, Any]) -> str:
"""Format quote data into a concise string.
Args:
quote_data: The response data from the Alpha Vantage Global Quote endpoint
Returns:
A formatted string containing the quote information
"""
try:
global_quote = quote_data.get("Global Quote", {})
if not global_quote:
return "No quote data available in the response"
return (
f"Price: ${global_quote.get('05. price', 'N/A')}\n"
f"Change: ${global_quote.get('09. change', 'N/A')} "
f"({global_quote.get('10. change percent', 'N/A')})\n"
f"Volume: {global_quote.get('06. volume', 'N/A')}\n"
f"High: ${global_quote.get('03. high', 'N/A')}\n"
f"Low: ${global_quote.get('04. low', 'N/A')}\n"
"---"
)
except Exception as e:
return f"Error formatting quote data: {str(e)}"
def format_company_info(overview_data: Dict[str, Any]) -> str:
"""Format company information into a concise string.
Args:
overview_data: The response data from the Alpha Vantage OVERVIEW endpoint
Returns:
A formatted string containing the company information
"""
try:
if not overview_data:
return "No company information available in the response"
return (
f"Name: {overview_data.get('Name', 'N/A')}\n"
f"Sector: {overview_data.get('Sector', 'N/A')}\n"
f"Industry: {overview_data.get('Industry', 'N/A')}\n"
f"Market Cap: ${overview_data.get('MarketCapitalization', 'N/A')}\n"
f"Description: {overview_data.get('Description', 'N/A')}\n"
f"Exchange: {overview_data.get('Exchange', 'N/A')}\n"
f"Currency: {overview_data.get('Currency', 'N/A')}\n"
"---"
)
except Exception as e:
return f"Error formatting company data: {str(e)}"
def format_crypto_rate(crypto_data: Dict[str, Any]) -> str:
"""Format cryptocurrency exchange rate data into a concise string.
Args:
crypto_data: The response data from the Alpha Vantage CURRENCY_EXCHANGE_RATE endpoint
Returns:
A formatted string containing the cryptocurrency exchange rate information
"""
try:
realtime_data = crypto_data.get("Realtime Currency Exchange Rate", {})
if not realtime_data:
return "No exchange rate data available in the response"
return (
f"From: {realtime_data.get('2. From_Currency Name', 'N/A')} ({realtime_data.get('1. From_Currency Code', 'N/A')})\n"
f"To: {realtime_data.get('4. To_Currency Name', 'N/A')} ({realtime_data.get('3. To_Currency Code', 'N/A')})\n"
f"Exchange Rate: {realtime_data.get('5. Exchange Rate', 'N/A')}\n"
f"Last Updated: {realtime_data.get('6. Last Refreshed', 'N/A')} {realtime_data.get('7. Time Zone', 'N/A')}\n"
f"Bid Price: {realtime_data.get('8. Bid Price', 'N/A')}\n"
f"Ask Price: {realtime_data.get('9. Ask Price', 'N/A')}\n"
"---"
)
except Exception as e:
return f"Error formatting cryptocurrency data: {str(e)}"
def format_time_series(time_series_data: Dict[str, Any], start_date: Optional[str] = None, end_date: Optional[str] = None, limit: int = 5) -> str:
"""Format time series data into a concise string with optional date filtering.
Args:
time_series_data: The response data from the Alpha Vantage TIME_SERIES_DAILY endpoint
start_date: Optional start date in YYYY-MM-DD format for filtering
end_date: Optional end date in YYYY-MM-DD format for filtering
limit: Number of data points to return when no date filtering is applied
Returns:
A formatted string containing the time series information
"""
try:
# Get the daily time series data
time_series = time_series_data.get("Time Series (Daily)", {})
if not time_series:
return "No time series data available in the response"
# Get metadata
metadata = time_series_data.get("Meta Data", {})
symbol = metadata.get("2. Symbol", "Unknown")
last_refreshed = metadata.get("3. Last Refreshed", "Unknown")
# Filter by date range if specified
filtered_data = {}
if start_date or end_date:
for date_str, values in time_series.items():
try:
date_obj = datetime.strptime(date_str, "%Y-%m-%d")
# Check start date filter
if start_date:
start_obj = datetime.strptime(start_date, "%Y-%m-%d")
if date_obj < start_obj:
continue
# Check end date filter
if end_date:
end_obj = datetime.strptime(end_date, "%Y-%m-%d")
if date_obj > end_obj:
continue
filtered_data[date_str] = values
except ValueError:
# Skip invalid date formats
continue
# Sort filtered data by date (most recent first)
sorted_items = sorted(filtered_data.items(), key=lambda x: x[0], reverse=True)
else:
# Use original data with limit
sorted_items = list(time_series.items())[:limit]
if not sorted_items:
return f"No time series data found for the specified date range"
# Build header
formatted_data = [
f"Time Series Data for {symbol} (Last Refreshed: {last_refreshed})\n"
]
# Add date range info if filtering was applied
if start_date or end_date:
date_range = ""
if start_date and end_date:
date_range = f"Date Range: {start_date} to {end_date}"
elif start_date:
date_range = f"From: {start_date}"
elif end_date:
date_range = f"Until: {end_date}"
formatted_data.append(f"{date_range} ({len(sorted_items)} data points)\n\n")
else:
formatted_data.append(f"(Showing {len(sorted_items)} most recent data points)\n\n")
# Format the data points
for date, values in sorted_items:
formatted_data.append(
f"Date: {date}\n"
f"Open: ${values.get('1. open', 'N/A')}\n"
f"High: ${values.get('2. high', 'N/A')}\n"
f"Low: ${values.get('3. low', 'N/A')}\n"
f"Close: ${values.get('4. close', 'N/A')}\n"
f"Volume: {values.get('5. volume', 'N/A')}\n"
"---\n"
)
return "\n".join(formatted_data)
except Exception as e:
return f"Error formatting time series data: {str(e)}"
def format_crypto_time_series(time_series_data: Dict[str, Any], series_type: str) -> str:
"""Format cryptocurrency time series data into a concise string.
Args:
time_series_data: The response data from Alpha Vantage Digital Currency endpoints
series_type: Type of time series (daily, weekly, monthly)
Returns:
A formatted string containing the cryptocurrency time series information
"""
try:
# Determine the time series key based on series_type
time_series_key = ""
if series_type == "daily":
time_series_key = "Time Series (Digital Currency Daily)"
elif series_type == "weekly":
time_series_key = "Time Series (Digital Currency Weekly)"
elif series_type == "monthly":
time_series_key = "Time Series (Digital Currency Monthly)"
else:
return f"Unknown series type: {series_type}"
# Get the time series data
time_series = time_series_data.get(time_series_key, {})
if not time_series:
all_keys = ", ".join(time_series_data.keys())
return f"No cryptocurrency time series data found with key: '{time_series_key}'.\nAvailable keys: {all_keys}"
# Get metadata
metadata = time_series_data.get("Meta Data", {})
crypto_symbol = metadata.get("2. Digital Currency Code", "Unknown")
crypto_name = metadata.get("3. Digital Currency Name", "Unknown")
market = metadata.get("4. Market Code", "Unknown")
market_name = metadata.get("5. Market Name", "Unknown")
last_refreshed = metadata.get("6. Last Refreshed", "Unknown")
time_zone = metadata.get("7. Time Zone", "Unknown")
# Format the header
formatted_data = [
f"{series_type.capitalize()} Time Series for {crypto_name} ({crypto_symbol})",
f"Market: {market_name} ({market})",
f"Last Refreshed: {last_refreshed} {time_zone}",
""
]
# Format the most recent 5 data points
for date, values in list(time_series.items())[:5]:
# Get price information - based on the API response, we now know the correct field names
open_price = values.get("1. open", "N/A")
high_price = values.get("2. high", "N/A")
low_price = values.get("3. low", "N/A")
close_price = values.get("4. close", "N/A")
volume = values.get("5. volume", "N/A")
formatted_data.append(f"Date: {date}")
formatted_data.append(f"Open: {open_price} {market}")
formatted_data.append(f"High: {high_price} {market}")
formatted_data.append(f"Low: {low_price} {market}")
formatted_data.append(f"Close: {close_price} {market}")
formatted_data.append(f"Volume: {volume}")
formatted_data.append("---")
return "\n".join(formatted_data)
except Exception as e:
return f"Error formatting cryptocurrency time series data: {str(e)}"
def format_earnings_calendar(earnings_data: List[Dict[str, str]], limit: int = 100, sort_by: str = "reportDate", sort_order: str = "desc") -> str:
"""Format earnings calendar data into a concise string with sorting.
Args:
earnings_data: List of earnings records from the Alpha Vantage EARNINGS_CALENDAR endpoint (CSV format)
limit: Number of earnings entries to display (default: 100)
sort_by: Field to sort by (default: reportDate)
sort_order: Sort order asc or desc (default: desc)
Returns:
A formatted string containing the sorted earnings calendar information
"""
try:
if not isinstance(earnings_data, list):
return f"Unexpected data format: {type(earnings_data)}"
if not earnings_data:
return "No earnings calendar data available"
# Sort the earnings data
def get_sort_key(earning):
value = earning.get(sort_by, "")
# Special handling for dates to ensure proper chronological sorting
if sort_by in ["reportDate", "fiscalDateEnding"]:
try:
# Convert date string to datetime for proper sorting
if value:
return datetime.strptime(value, "%Y-%m-%d")
else:
return datetime.min # Put empty dates at the beginning
except ValueError:
return datetime.min
# Special handling for numeric fields like estimate
elif sort_by == "estimate":
try:
if value and value.strip():
return float(value)
else:
return 0.0
except ValueError:
return 0.0
# For text fields (symbol, name), return as-is for alphabetical sorting
else:
return str(value).upper()
sorted_earnings = sorted(
earnings_data,
key=get_sort_key,
reverse=(sort_order == "desc")
)
formatted = [f"Upcoming Earnings Calendar (Sorted by {sort_by} {sort_order}):\n\n"]
# Display limited number of entries
display_earnings = sorted_earnings[:limit] if limit > 0 else sorted_earnings
for earning in display_earnings:
symbol = earning.get('symbol', 'N/A')
name = earning.get('name', 'N/A')
report_date = earning.get('reportDate', 'N/A')
fiscal_date = earning.get('fiscalDateEnding', 'N/A')
estimate = earning.get('estimate', 'N/A')
currency = earning.get('currency', 'N/A')
formatted.append(f"Company: {symbol} - {name}\n")
formatted.append(f"Report Date: {report_date}\n")
formatted.append(f"Fiscal Date End: {fiscal_date}\n")
# Format estimate nicely
if estimate and estimate != 'N/A' and estimate.strip():
try:
est_float = float(estimate)
formatted.append(f"Estimate: ${est_float:.2f} {currency}\n")
except ValueError:
formatted.append(f"Estimate: {estimate} {currency}\n")
else:
formatted.append(f"Estimate: Not available\n")
formatted.append("---\n")
if limit > 0 and len(earnings_data) > limit:
formatted.append(f"\n... and {len(earnings_data) - limit} more earnings reports")
return "".join(formatted)
except Exception as e:
return f"Error formatting earnings calendar data: {str(e)}"
def format_historical_earnings(earnings_data: Dict[str, Any], limit_annual: int = 5, limit_quarterly: int = 8) -> str:
"""Format historical earnings data into a concise string.
Args:
earnings_data: The response data from the Alpha Vantage EARNINGS endpoint
limit_annual: Number of annual earnings to display (default: 5)
limit_quarterly: Number of quarterly earnings to display (default: 8)
Returns:
A formatted string containing the historical earnings information
"""
try:
if "Error Message" in earnings_data:
return f"Error: {earnings_data['Error Message']}"
symbol = earnings_data.get("symbol", "Unknown")
formatted = [f"Historical Earnings for {symbol}:\n\n"]
# Format Annual Earnings
annual_earnings = earnings_data.get("annualEarnings", [])
if annual_earnings:
formatted.append("=== ANNUAL EARNINGS ===\n")
display_annual = annual_earnings[:limit_annual] if limit_annual > 0 else annual_earnings
for earning in display_annual:
fiscal_date = earning.get("fiscalDateEnding", "N/A")
reported_eps = earning.get("reportedEPS", "N/A")
formatted.append(f"Fiscal Year End: {fiscal_date}\n")
formatted.append(f"Reported EPS: ${reported_eps}\n")
formatted.append("---\n")
if limit_annual > 0 and len(annual_earnings) > limit_annual:
formatted.append(f"... and {len(annual_earnings) - limit_annual} more annual reports\n")
formatted.append("\n")
# Format Quarterly Earnings
quarterly_earnings = earnings_data.get("quarterlyEarnings", [])
if quarterly_earnings:
formatted.append("=== QUARTERLY EARNINGS ===\n")
display_quarterly = quarterly_earnings[:limit_quarterly] if limit_quarterly > 0 else quarterly_earnings
for earning in display_quarterly:
fiscal_date = earning.get("fiscalDateEnding", "N/A")
reported_date = earning.get("reportedDate", "N/A")
reported_eps = earning.get("reportedEPS", "N/A")
estimated_eps = earning.get("estimatedEPS", "N/A")
surprise = earning.get("surprise", "N/A")
surprise_pct = earning.get("surprisePercentage", "N/A")
report_time = earning.get("reportTime", "N/A")
formatted.append(f"Fiscal Quarter End: {fiscal_date}\n")
formatted.append(f"Reported Date: {reported_date}\n")
formatted.append(f"Reported EPS: ${reported_eps}\n")
formatted.append(f"Estimated EPS: ${estimated_eps}\n")
# Format surprise with proper handling
if surprise != "N/A" and surprise_pct != "N/A":
try:
surprise_float = float(surprise)
surprise_pct_float = float(surprise_pct)
if surprise_float >= 0:
formatted.append(f"Surprise: +${surprise_float:.2f} (+{surprise_pct_float:.2f}%)\n")
else:
formatted.append(f"Surprise: ${surprise_float:.2f} ({surprise_pct_float:.2f}%)\n")
except ValueError:
formatted.append(f"Surprise: {surprise} ({surprise_pct}%)\n")
else:
formatted.append(f"Surprise: {surprise} ({surprise_pct}%)\n")
formatted.append(f"Report Time: {report_time}\n")
formatted.append("---\n")
if limit_quarterly > 0 and len(quarterly_earnings) > limit_quarterly:
formatted.append(f"... and {len(quarterly_earnings) - limit_quarterly} more quarterly reports\n")
if not annual_earnings and not quarterly_earnings:
formatted.append("No historical earnings data available\n")
return "".join(formatted)
except Exception as e:
return f"Error formatting historical earnings data: {str(e)}"
def format_historical_options(
options_data: Dict[str, Any],
limit: int = 10,
sort_by: str = "strike",
sort_order: str = "asc",
expiry_date: Optional[str] = None,
min_strike: Optional[float] = None,
max_strike: Optional[float] = None,
contract_id: Optional[str] = None,
contract_type: Optional[str] = None
) -> str:
"""Format historical options chain data into a concise string with advanced filtering and sorting.
Args:
options_data: The response data from the Alpha Vantage HISTORICAL_OPTIONS endpoint
limit: Number of contracts to return after filtering (-1 for all)
sort_by: Field to sort by
sort_order: Sort order (asc or desc)
expiry_date: Optional expiration date filter (YYYY-MM-DD)
min_strike: Optional minimum strike price filter
max_strike: Optional maximum strike price filter
contract_id: Optional specific contract ID filter
contract_type: Optional contract type filter (call/put/C/P)
Returns:
A formatted string containing the filtered and sorted historical options information
"""
try:
if "Error Message" in options_data:
return f"Error: {options_data['Error Message']}"
options_chain = options_data.get("data", [])
if not options_chain:
return "No options data available in the response"
# Apply filters
filtered_chain = []
for contract in options_chain:
# Contract ID filter (exact match)
if contract_id and contract.get('contractID', '') != contract_id:
continue
# Expiry date filter
if expiry_date:
contract_expiry = contract.get('expiration', '')
if contract_expiry != expiry_date:
continue
# Strike price filters
if min_strike is not None or max_strike is not None:
try:
strike_str = str(contract.get('strike', '0')).replace('$', '').strip()
if strike_str:
strike_price = float(strike_str)
if min_strike is not None and strike_price < min_strike:
continue
if max_strike is not None and strike_price > max_strike:
continue
except (ValueError, TypeError):
continue
# Contract type filter
if contract_type:
contract_type_norm = contract_type.upper()
# Handle both full names and single letters
if contract_type_norm in ['CALL', 'C']:
expected_types = ['call', 'C', 'CALL']
elif contract_type_norm in ['PUT', 'P']:
expected_types = ['put', 'P', 'PUT']
else:
expected_types = [contract_type]
actual_type = contract.get('type', '')
if actual_type not in expected_types:
continue
filtered_chain.append(contract)
if not filtered_chain:
filters_applied = []
if contract_id:
filters_applied.append(f"contract_id={contract_id}")
if expiry_date:
filters_applied.append(f"expiry={expiry_date}")
if min_strike is not None:
filters_applied.append(f"min_strike={min_strike}")
if max_strike is not None:
filters_applied.append(f"max_strike={max_strike}")
if contract_type:
filters_applied.append(f"type={contract_type}")
filter_text = ", ".join(filters_applied) if filters_applied else "applied"
return f"No options contracts found matching the specified filters: {filter_text}"
formatted = [
f"Historical Options Data (Filtered):\n",
f"Status: {options_data.get('message', 'N/A')}\n",
]
# Add filter summary
filters_applied = []
if contract_id:
filters_applied.append(f"Contract ID: {contract_id}")
if expiry_date:
filters_applied.append(f"Expiry: {expiry_date}")
if min_strike is not None or max_strike is not None:
strike_range = []
if min_strike is not None:
strike_range.append(f"min ${min_strike}")
if max_strike is not None:
strike_range.append(f"max ${max_strike}")
filters_applied.append(f"Strike: {' - '.join(strike_range)}")
if contract_type:
filters_applied.append(f"Type: {contract_type}")
if filters_applied:
formatted.append(f"Filters: {', '.join(filters_applied)}\n")
formatted.append(f"Found {len(filtered_chain)} contracts, sorted by: {sort_by} ({sort_order})\n\n")
# Convert string values to float for numeric sorting
def get_sort_key(contract):
value = contract.get(sort_by, 0)
try:
# Remove $ and % signs if present
if isinstance(value, str):
value = value.replace('$', '').replace('%', '')
return float(value)
except (ValueError, TypeError):
return value
# Sort the filtered chain
sorted_chain = sorted(
filtered_chain,
key=get_sort_key,
reverse=(sort_order == "desc")
)
# If limit is -1, show all contracts
display_contracts = sorted_chain if limit == -1 else sorted_chain[:limit]
for contract in display_contracts:
formatted.append(f"Contract Details:\n")
formatted.append(f"Contract ID: {contract.get('contractID', 'N/A')}\n")
formatted.append(f"Expiration: {contract.get('expiration', 'N/A')}\n")
formatted.append(f"Strike: ${contract.get('strike', 'N/A')}\n")
formatted.append(f"Type: {contract.get('type', 'N/A')}\n")
formatted.append(f"Last: ${contract.get('last', 'N/A')}\n")
formatted.append(f"Mark: ${contract.get('mark', 'N/A')}\n")
formatted.append(f"Bid: ${contract.get('bid', 'N/A')} (Size: {contract.get('bid_size', 'N/A')})\n")
formatted.append(f"Ask: ${contract.get('ask', 'N/A')} (Size: {contract.get('ask_size', 'N/A')})\n")
formatted.append(f"Volume: {contract.get('volume', 'N/A')}\n")
formatted.append(f"Open Interest: {contract.get('open_interest', 'N/A')}\n")
formatted.append(f"IV: {contract.get('implied_volatility', 'N/A')}\n")
formatted.append(f"Delta: {contract.get('delta', 'N/A')}\n")
formatted.append(f"Gamma: {contract.get('gamma', 'N/A')}\n")
formatted.append(f"Theta: {contract.get('theta', 'N/A')}\n")
formatted.append(f"Vega: {contract.get('vega', 'N/A')}\n")
formatted.append(f"Rho: {contract.get('rho', 'N/A')}\n")
formatted.append("---\n")
if limit != -1 and len(sorted_chain) > limit:
formatted.append(f"\n... and {len(sorted_chain) - limit} more contracts")
return "".join(formatted)
except Exception as e:
return f"Error formatting options data: {str(e)}"
def format_realtime_options(options_data: Dict[str, Any]) -> str:
"""Format realtime options data into a concise string.
Args:
options_data: The response data from the Alpha Vantage REALTIME_OPTIONS endpoint
Returns:
A formatted string containing the realtime options information
"""
try:
if "Error Message" in options_data:
return f"Error: {options_data['Error Message']}"
# Get the options contracts list
options_list = options_data.get("data", [])
if not options_list:
return "No realtime options data available in the response"
# Detect placeholder/demo data patterns
def is_placeholder_data(contracts_list):
"""Detect if the response contains placeholder/demo data"""
for contract in contracts_list:
symbol = contract.get("symbol", "")
contract_id = contract.get("contractID", "")
expiration = contract.get("expiration", "")
# Check for common placeholder patterns
if (symbol == "XXYYZZ" or
"XXYYZZ" in contract_id or
expiration == "2099-99-99" or
"999999" in contract_id):
return True
return False
if is_placeholder_data(options_list):
# Check if we're using demo API key
api_key_status = "demo API key" if API_KEY == "demo" else f"API key ending in ...{API_KEY[-4:]}" if API_KEY and len(API_KEY) > 4 else "no API key set"
return (
"❌ PREMIUM FEATURE REQUIRED ❌\n\n"
"The realtime options data you requested requires a premium Alpha Vantage subscription.\n"
"The API returned placeholder/demo data instead of real market data.\n\n"
f"Current API key status: {api_key_status}\n\n"
"Possible causes:\n"
"1. Using demo API key instead of your actual API key\n"
"2. API key is valid but account doesn't have premium access\n"
"3. Alpha Vantage returns demo data for free accounts on this endpoint\n\n"
"Solutions:\n"
"1. Ensure your actual API key is set in ALPHA_VANTAGE_API_KEY environment variable\n"
"2. Upgrade to Alpha Vantage Premium (600 or 1200 requests/minute plan)\n"
"3. Use 'get-historical-options' tool for historical data (available with free accounts)\n\n"
"Learn more: https://www.alphavantage.co/premium/\n\n"
"Note: Historical options data may meet your analysis needs and works with free accounts."
)
# Group contracts by expiration date and then by strike price
contracts_by_expiry = {}
for contract in options_list:
expiry = contract.get("expiration", "Unknown")
strike = contract.get("strike", "0.00")
contract_type = contract.get("type", "unknown")
if expiry not in contracts_by_expiry:
contracts_by_expiry[expiry] = {}
if strike not in contracts_by_expiry[expiry]:
contracts_by_expiry[expiry][strike] = {}
contracts_by_expiry[expiry][strike][contract_type] = contract
# Extract symbol from first contract
symbol = options_list[0].get("symbol", "Unknown") if options_list else "Unknown"
formatted = [
f"Realtime Options Data for {symbol}\n",
f"Found {len(options_list)} contracts\n\n"
]
# Sort by expiration dates
sorted_expiries = sorted(contracts_by_expiry.keys())
for expiry in sorted_expiries:
formatted.append(f"=== Expiration: {expiry} ===\n")
# Sort strikes numerically
strikes = contracts_by_expiry[expiry]
sorted_strikes = sorted(strikes.keys(), key=lambda x: float(x) if str(x).replace('.', '').isdigit() else 0)
for strike in sorted_strikes:
contract_types = strikes[strike]
for contract_type in ["call", "put"]:
if contract_type in contract_types:
contract = contract_types[contract_type]
formatted.append(f"\nStrike: ${strike} ({contract_type.upper()})\n")
formatted.append(f"Contract ID: {contract.get('contractID', 'N/A')}\n")
formatted.append(f"Last: ${contract.get('last', 'N/A')}\n")
formatted.append(f"Mark: ${contract.get('mark', 'N/A')}\n")
formatted.append(f"Bid: ${contract.get('bid', 'N/A')} (Size: {contract.get('bid_size', 'N/A')})\n")
formatted.append(f"Ask: ${contract.get('ask', 'N/A')} (Size: {contract.get('ask_size', 'N/A')})\n")
formatted.append(f"Volume: {contract.get('volume', 'N/A')}\n")
formatted.append(f"Open Interest: {contract.get('open_interest', 'N/A')}\n")
formatted.append(f"Date: {contract.get('date', 'N/A')}\n")
# Include Greeks if available
if 'implied_volatility' in contract:
formatted.append(f"IV: {contract.get('implied_volatility', 'N/A')}\n")
if 'delta' in contract:
formatted.append(f"Delta: {contract.get('delta', 'N/A')}\n")
if 'gamma' in contract:
formatted.append(f"Gamma: {contract.get('gamma', 'N/A')}\n")
if 'theta' in contract:
formatted.append(f"Theta: {contract.get('theta', 'N/A')}\n")
if 'vega' in contract:
formatted.append(f"Vega: {contract.get('vega', 'N/A')}\n")
if 'rho' in contract:
formatted.append(f"Rho: {contract.get('rho', 'N/A')}\n")
formatted.append("---\n")
formatted.append("\n")
return "".join(formatted)
except Exception as e:
return f"Error formatting realtime options data: {str(e)}"
def format_etf_profile(etf_data: Dict[str, Any]) -> str:
"""Format ETF profile data into a concise string.
Args:
etf_data: The response data from the Alpha Vantage ETF_PROFILE endpoint
Returns:
A formatted string containing the ETF profile information
"""
try:
if "Error Message" in etf_data:
return f"Error: {etf_data['Error Message']}"
if not etf_data:
return "No ETF profile data available in the response"
# Extract basic ETF information
net_assets = etf_data.get("net_assets", "N/A")
net_expense_ratio = etf_data.get("net_expense_ratio", "N/A")
portfolio_turnover = etf_data.get("portfolio_turnover", "N/A")
dividend_yield = etf_data.get("dividend_yield", "N/A")
inception_date = etf_data.get("inception_date", "N/A")
leveraged = etf_data.get("leveraged", "N/A")
formatted = [
f"ETF Profile\n\n",
f"Basic Information:\n"
]
# Format net assets
if net_assets != "N/A" and net_assets.replace('.', '').isdigit():
formatted.append(f"Net Assets: ${float(net_assets):,.0f}\n")
else:
formatted.append(f"Net Assets: {net_assets}\n")
# Format net expense ratio
if net_expense_ratio != "N/A" and net_expense_ratio.replace('.', '').replace('-', '').isdigit():
formatted.append(f"Net Expense Ratio: {float(net_expense_ratio):.3%}\n")
else:
formatted.append(f"Net Expense Ratio: {net_expense_ratio}\n")
# Format portfolio turnover
if portfolio_turnover != "N/A" and portfolio_turnover.replace('.', '').replace('-', '').isdigit():
formatted.append(f"Portfolio Turnover: {float(portfolio_turnover):.1%}\n")
else:
formatted.append(f"Portfolio Turnover: {portfolio_turnover}\n")
# Format dividend yield
if dividend_yield != "N/A" and dividend_yield.replace('.', '').replace('-', '').isdigit():
formatted.append(f"Dividend Yield: {float(dividend_yield):.2%}\n")
else:
formatted.append(f"Dividend Yield: {dividend_yield}\n")
formatted.extend([
f"Inception Date: {inception_date}\n",
f"Leveraged: {leveraged}\n\n"
])
# Format sectors if available
sectors = etf_data.get("sectors", [])
if sectors:
formatted.append("Sector Allocation:\n")
for sector in sectors:
sector_name = sector.get("sector", "Unknown")
weight = sector.get("weight", "0")
try:
weight_pct = float(weight) * 100
formatted.append(f"{sector_name}: {weight_pct:.1f}%\n")
except (ValueError, TypeError):
formatted.append(f"{sector_name}: {weight}\n")
formatted.append("\n")
# Format top holdings if available
holdings = etf_data.get("holdings", [])
if holdings:
formatted.append("Top Holdings:\n")
# Show top 10 holdings
for i, holding in enumerate(holdings[:10]):
symbol = holding.get("symbol", "N/A")
description = holding.get("description", "N/A")
weight = holding.get("weight", "0")
try:
weight_pct = float(weight) * 100
formatted.append(f"{i+1:2d}. {symbol} - {description}: {weight_pct:.2f}%\n")
except (ValueError, TypeError):
formatted.append(f"{i+1:2d}. {symbol} - {description}: {weight}\n")
if len(holdings) > 10:
formatted.append(f"\n... and {len(holdings) - 10} more holdings\n")
formatted.append(f"\nTotal Holdings: {len(holdings)}\n")
return "".join(formatted)
except Exception as e:
return f"Error formatting ETF profile data: {str(e)}"
```