Exclusive Deal for Polygon Readers: Dive into Blue Prince with Special Offer!

In this practical guide, we’ll create an MCP (Model Context Protocol) server enabling Claude Desktop to obtain stock market news sentiment and daily top gainers and movers using the AlphaVantage API. Since most large language models cannot directly retrieve real-time financial data, this approach empowers real-time insights through MCP.

We’ll introduce two functionalities from our server:

• get_news_sentiment
• get_top_movers

Let’s explore each step.

Step 1: Setting Up the Environment

We’ll start by configuring the environment and installing the uv package manager. For Mac or Linux:

curl -LsSf https://astral.sh/uv/install.sh | sh

For Windows (PowerShell):

powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

Proceed to create a new project directory and initialize it with uv:

uv init stockNews
cd stockNews

Create and activate a virtual environment. For Mac or Linux:

uv venv
source .venv/bin/activate

For Windows:

uv venv
.venvScriptsactivate

Next, install the necessary dependencies:

uv add mcp httpx python-dotenv

Step 3: Setting Up the Environment Variables

Create a .env file to include the AlphaVantage API key. To acquire a free API key, follow the instructions on their website. Then, add this line in your .env file:

ALPHA_VANTAGE_API_KEY=your_api_key

Step 4: Implementing the MCP Server and Integrating AlphaVantage

Create a stockNews.py file in the designated directory and insert the following code snippets:

Import necessary packages and configure the instance:

from typing import Any
import os
import httpx
from mcp.server.fastmcp import FastMCP
from dotenv import load_dotenv

# Load .env variables
load_dotenv()
API_KEY = os.getenv("ALPHA_VANTAGE_API_KEY")

# Initialize FastMCP server
mcp = FastMCP("alpha-finance")

# Constants
BASE_URL = "https://www.alphavantage.co/query"

Helper Functions

Add helper functions for data queries from AlphaVantage:

async def call_alpha_vantage(endpoint: str, params: dict[str, Any]) -> dict[str, Any] | None:
    """Generic async caller to Alpha Vantage."""
    params["apikey"] = API_KEY
    params["function"] = endpoint
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(BASE_URL, params=params, timeout=30.0)
            response.raise_for_status()
            return response.json()
        except Exception:
            return None

Implementing Tool Execution

Define the tool execution handler:

@mcp.tool()
async def get_news_sentiment(ticker: str) -> str:
    """Get news sentiment data for a stock ticker.

    Args:
        ticker: Stock ticker symbol (e.g., MSFT, AAPL)
    """
    data = await call_alpha_vantage("NEWS_SENTIMENT", {"tickers": ticker.upper()})
    if not data or "feed" not in data:
        return "Couldn't retrieve news sentiment."

    articles = data["feed"][:3]
    result = []
    for item in articles:
        result.append(f"""
📰 {item['title']}
Summary: {item['summary']}
Source: {item['source']} | Published: {item['time_published']}
""")
    return "n---n".join(result)

@mcp.tool()
async def get_top_movers() -> str:
    """Get top gainers and losers from the stock market.

    No arguments required.
    """
    data = await call_alpha_vantage("TOP_GAINERS_LOSERS", {})
    if not data:
        return "Couldn't retrieve top movers."

    gainers = data.get("top_gainers", [])[:3]
    losers = data.get("top_losers", [])[:3]

    result = "**Top Gainers**n"
    result += "n".join([
        f"{g['ticker']} ({g.get('change_percentage', 'N/A')})"
        for g in gainers
    ])

    result += "nn**Top Losers**n"
    result += "n".join([
        f"{l['ticker']} ({l.get('change_percentage', 'N/A')})"
        for l in losers
    ])

    return result

Running the Server

Initialize and run the server:

if __name__ == "__main__":
    mcp.run(transport="stdio")

Step 5: Testing the Server

Ensure you