itinai content

←back to Blog

Implementing OAuth 2.1 for MCP Servers with Scalekit: A Step-by-Step Coding Tutorial

«`html

Implementing OAuth 2.1 for MCP Servers with Scalekit: A Step-by-Step Coding Tutorial

In this tutorial, we’ll explore how to implement OAuth 2.1 for MCP servers step by step. To keep things practical, we’ll build a simple finance sentiment analysis server and secure it using Scalekit, a tool that simplifies the OAuth setup process.

With Scalekit, all we need to do is expose a metadata endpoint URL for MCP clients to discover the server and add authorization middleware for secure token-based authentication. Scalekit manages all the complex OAuth 2.1 flows behind the scenes, eliminating the need for manual token generation, refresh, or validation. Once this setup is complete, your MCP server is ready to handle authenticated requests seamlessly.

Setting Up Dependencies

Alpha Vantage API

To fetch stock news sentiment, we’ll use the Alpha Vantage API. To get a free API key:

  • Visit the Alpha Vantage platform.
  • Enter your email and the required details.
  • You’ll receive your API key—copy it and store it securely, as you’ll need it to authenticate your requests.

Node.js

To run the MCP Inspector for testing our application, we need Node.js installed.

  • Download the latest version of Node.js from nodejs.org.
  • Run the installer.
  • Keep the default settings and complete the installation.

Python Dependencies

Install the required Python packages:

pip install fastapi fastmcp mcp scalekit-sdk-python

Scalekit

To start using Scalekit, follow these steps:

  • Create Your Scalekit Account: Go to scalekit.com and sign up. Scalekit offers a free tier, so you don’t need to worry about billing.
  • Once signed in, click “Activate Full-Stack Auth.”
  • Set Up Permissions: Open the Authorization panel and click “Add Permission.” Use the following values:
  • Permission Name: news:read
  • Description: Use Alpha Vantage to get Stock Sentiment

Permissions in Scalekit define and manage scopes that control what features or resources your application can access. For example, the news:read permission allows your MCP server to access stock sentiment data from Alpha Vantage.

Add Your MCP Server

Go to the MCP Servers section and click “Add MCP Server.” Fill in the required fields:

  • Server Name: Any name you prefer.
  • Resource Identifier: A unique identifier for your MCP server. This value is included in the aud claim of access tokens, helping the server validate requests.

For local testing, set it as:

http://localhost:10000/mcp/

Set the scope to the permission you just created: news:read.

Get API Credentials

Go to Settings → API Credentials. Copy your Client ID and Environment URL. Click Generate New Secret to create your Secret Key. Store these values securely, as we’ll need them later for configuration.

Configuration File (config.py)

We will create a config file to load all the environment variables which will be used later:

import os
from dotenv import load_dotenv

load_dotenv()

class Settings():
    ALPHA_VANTAGE_API_KEY = os.environ.get('ALPHA_VANTAGE_API_KEY')
    METADATA_JSON_RESPONSE = os.environ.get('METADATA_JSON_RESPONSE')
    SCALEKIT_ENVIRONMENT_URL = os.environ.get('SCALEKIT_ENVIRONMENT_URL')
    SCALEKIT_CLIENT_ID = os.environ.get('SCALEKIT_CLIENT_ID')
    SCALEKIT_CLIENT_SECRET = os.environ.get('SCALEKIT_CLIENT_SECRET')
    SCALEKIT_RESOURCE_METADATA_URL = os.environ.get('SCALEKIT_RESOURCE_METADATA_URL')
    SCALEKIT_AUTHORIZATION_SERVERS = os.environ.get('SCALEKIT_AUTHORIZATION_SERVERS')
    SCALEKIT_AUDIENCE_NAME = os.environ.get('SCALEKIT_AUDIENCE_NAME')
    SCALEKIT_RESOURCE_NAME = os.environ.get('SCALEKIT_RESOURCE_NAME')
    SCALEKIT_RESOURCE_DOCS_URL = os.environ.get('SCALEKIT_RESOURCE_DOCS_URL')
    PORT = 10000

settings = Settings()

Stock Sentiment Logic (finance.py)

This code block fetches real-time news sentiment data for a given stock ticker using the Alpha Vantage API. It retrieves the top three recent articles, summarizing their title, summary, source, and publication time for quick insights:

from mcp.server.fastmcp import FastMCP
from typing import Any
import os
import httpx
from typing import Dict, List
from config import settings

mcp = FastMCP("finance-news")

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

async def call_alpha_vantage(endpoint: str, params: dict[str, Any]) -> dict[str, Any] | None:
    params["apikey"] = settings.ALPHA_VANTAGE_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

@mcp.tool()
async def get_news_sentiment(ticker: str) -> str:
    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)

Authorization Middleware

This middleware acts as an authorization layer for your MCP server, ensuring that only authenticated requests are processed. It uses the Scalekit client to validate access tokens on every incoming request. If the request isn’t for a public path, it looks for an Authorization header with a valid Bearer token. If the token is missing, invalid, or expired, the middleware responds with a 401 Unauthorized error.

Logging is integrated throughout the process to capture key events, making it easier to debug and audit authentication flows. This middleware will be imported and added to the server file to protect all secure endpoints:

import json
import logging
from fastapi import HTTPException, Request
from fastapi.security import HTTPBearer
from fastapi.responses import JSONResponse
from scalekit import ScalekitClient
from starlette.middleware.base import BaseHTTPMiddleware

from config import settings

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

security = HTTPBearer()

scalekit_client = ScalekitClient(
    settings.SCALEKIT_ENVIRONMENT_URL,
    settings.SCALEKIT_CLIENT_ID,
    settings.SCALEKIT_CLIENT_SECRET
)

class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        if request.url.path.startswith("/.well-known/"):
            return await call_next(request)

        try:
            auth_header = request.headers.get("Authorization")
            if not auth_header or not auth_header.startswith("Bearer "):
                raise HTTPException(status_code=401, detail="Missing or invalid authorization header")

            token = auth_header.split(" ")[1]

            request_body = await request.body()
            
            try:
                request_data = json.loads(request_body.decode('utf-8'))
            except (json.JSONDecodeError, UnicodeDecodeError):
                request_data = {}
            
            try:
                scalekit_client.validate_access_token(token)
                
            except Exception as e:
                raise HTTPException(status_code=401, detail="Token validation failed")

        except HTTPException as e:
            return JSONResponse(
                status_code=e.status_code,
                content={"error": "unauthorized" if e.status_code == 401 else "forbidden", "error_description": e.detail},
                headers={
                    "WWW-Authenticate": f'Bearer realm="OAuth", resource_metadata="{settings.SCALEKIT_RESOURCE_METADATA_URL}"'
                }
            )

        return await call_next(request)

MCP Server (server.py)

This script sets up a FastAPI application integrated with an MCP server for stock news sentiment analysis. It begins by importing the necessary libraries, including FastAPI, CORS middleware, and a custom authentication middleware:

import contextlib
import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
import json
from auth import AuthMiddleware
from config import settings
from finance import mcp as finance_news_server

@contextlib.asynccontextmanager
async def lifespan(app: FastAPI):
    async with finance_news_server.session_manager.run():
        yield

app = FastAPI(lifespan=lifespan)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
    allow_headers=["*"],
)

@app.get("/.well-known/oauth-protected-resource/mcp")
async def oauth_protected_resource_metadata():
    return {
        "authorization_servers": [settings.SCALEKIT_AUTHORIZATION_SERVERS],
        "bearer_methods_supported": ["header"],
        "resource": settings.SCALEKIT_RESOURCE_NAME,
        "resource_documentation": settings.SCALEKIT_RESOURCE_DOCS_URL,
        "scopes_supported": [
          "mcp:tools:news:read"
        ],
    }

mcp_server = finance_news_server.streamable_http_app()
app.add_middleware(AuthMiddleware)
app.mount("/", mcp_server)

def main():
    uvicorn.run(app, host="localhost", port=settings.PORT, log_level="debug")

if __name__ == "__main__":
    main()

Running the Server

To run the server, execute python server.py, which will start the application on localhost:10000. To test the setup, open another terminal and run:

npx @modelcontextprotocol/inspector

Once the MCP Inspector is running, enter http://localhost:10000/mcp as the server URL. If you attempt to connect without providing valid credentials, you will encounter a connection error. Provide the Bearer token using the secret ID you generated in Scalekit to authenticate successfully.

Feel free to check out our GitHub Page for Tutorials, Codes, and Notebooks. Also, follow us on Twitter and join our 100k+ ML SubReddit.

«`