Implementación de OAuth 2.1 para servidores MCP con scalekit: un tutorial de codificación paso a paso

En este tutorial, exploraremos cómo implementar OAuth 2.1 para MCP servidors paso a paso. Para mantener las cosas prácticas, construiremos un servidor de análisis de sentimientos financiero simple y lo aseguraremos usando Scalekit, una herramienta que hace que la configuración de OAuth sea más rápida y más fácil.

Con ScaleKit, todo lo que necesitamos hacer es exponer una URL de punto final de metadatos para que los clientes de MCP descubran el servidor y agregue Autorización Middleware para la autenticación segura basada en token. ScaleKit maneja todos los flujos complejos de OAuth 2.1 detrás de escena, por lo que no necesita implementar o administrar manualmente la generación de tokens, actualizar o validación. Una vez que esta configuración está completa, su servidor MCP está listo para manejar las solicitudes autenticadas sin problemas. Mira el Códigos completos aquí.

API alfa Vantage

Para obtener un sentimiento de noticias de valores, usaremos la API Alpha Vantage. Para obtener una clave API gratuita:

• Visite la plataforma Alpha Vantage utilizando esta enlace
• Ingrese su correo electrónico y los detalles requeridos.
• Recibirá su clave API: ciféla y almacéntela de forma segura, ya que la necesitará para autenticar sus solicitudes.

Nodo JS

Para ejecutar el Inspector de MCP Para probar nuestra aplicación, necesitamos Node.js instalados.

• Descargue la última versión de Node.js desde nodejs.org
• Ejecute el instalador.
• Mantenga la configuración predeterminada y complete la instalación.

Dependencias de Python

pip install fastapi fastmcp mcp scalekit-sdk-python

Kit de escala

Para comenzar a usar Kit de escalasiga estos pasos:

Crea tu cuenta de kit de escala

• Ir a scalekit.com y regístrese.
• ScaleKit ofrece un nivel gratuito, por lo que no necesita preocuparse por la facturación.
• Una vez iniciado, haga clic en “Activar autenticación de stats. “

Configurar permisos

• Abra el panel de autorización.
• En la sección Permisos, haga clic en “Agregar permiso”.
• Use los siguientes valores:

Nombre del permiso: Noticias: Leer

Descripción: Use Alpha Vantage para obtener el sentimiento de stock

Los permisos en el kit de escala se utilizan para definir y administrar ámbitos que controlan a qué características o recursos puede acceder su aplicación. Por ejemplo, el Noticias: Leer El permiso permite que su servidor MCP acceda a los datos de sentimientos de stock de Alpha Vantage, mientras que se pueden crear otros permisos para compilar características o API adicionales dentro de su aplicación.

Agregue su servidor MCP

• Vaya a la sección Servidores MCP y haga clic en “Agregar servidor MCP”.
• Complete los campos requeridos:

Nombre del servidor: Cualquier nombre que prefiera.

Identificador de recursos: Un identificador único para su servidor MCP. Este valor se incluye en el reclamo AUD de los tokens de acceso, ayudando al servidor a validar las solicitudes.

Para las pruebas locales, configúrelo como:

http://localhost:10000/mcp/

Al usar FastMCP, la ruta /MCP se agrega automáticamente al punto final. Asegúrese de incluir la barra de arrastre al final para evitar problemas de configuración. Mira el Códigos completos aquí.

Establezca el alcance en el permiso que acaba de crear: Noticias: Leer

Una vez que se crea el servidor, ScaleKit generará sus metadatos de recursos. Asegúrese de anotar el identificador del servidor MCP (que se encuentra junto al nombre del servidor, por ejemplo, res_88056357768398086), como lo necesitarás más tarde.

Ejemplo de metadatos de recursos

Sus metadatos se verán similares a esto (pero únicos para su cuenta):

URL de punto final de metadatos:

/.well-known/oauth-protected-resource/mcp

Metadatos de recursos JSON:

{
  "authorization_servers": [
    "https://zapp.scalekit.dev/resources/res_88056357768398086"
  ],
  "bearer_methods_supported": ["header"],
  "resource": "http://localhost:10000/mcp/",
  "resource_documentation": "http://localhost:10000/mcp/docs",
  "scopes_supported": ["news:read"]
}

Obtener credenciales de API

• Vaya a Configuración → Credenciales de API.
• Copie su ID de cliente y URL de entorno.
• Haga clic en Generar un nuevo secreto para crear su clave secreta.

Almacene estos valores de forma segura: los necesitaremos más tarde para la configuración.

.env

Ahora crearemos un archivo .env con las siguientes variables

ALPHA_VANTAGE_API_KEY=<YOUR_ALPHA_VANTAGE_API_KEY>
METADATA_JSON_RESPONSE=<YOUR_METADATA_JSON_RESPONSE>

SCALEKIT_ENVIRONMENT_URL=<YOUR_SCALEKIT_ENVIRONMENT_URL>
SCALEKIT_CLIENT_ID=<YOUR_SCALEKIT_CLIENT_ID>
SCALEKIT_CLIENT_SECRET=<YOUR_SCALEKIT_CLIENT_SECRET>
SCALEKIT_RESOURCE_METADATA_URL=<YOUR_SCALEKIT_RESOURCE_METADATA_URL>
SCALEKIT_AUTHORIZATION_SERVERS=<YOUR_SCALEKIT_AUTHORIZATION_SERVERS>
SCALEKIT_AUDIENCE_NAME=<YOUR_SCALEKIT_AUDIENCE_NAME>
SCALEKIT_RESOUCE_NAME=<YOUR_SCALEKIT_RESOURCE_NAME>
SCALEKIT_RESOUCE_DOCS_URL=<YOUR_SCALEKIT_RESOURCE_DOCS_URL>

Alfa_vantage_api_key

Su clave API personal de Alpha Vantage, utilizada para obtener datos de sentimientos de stock.

Metadata_json_response

La respuesta JSON generada por ScaleKit cuando configura su servidor MCP.

Contiene detalles como servidores de autorización, ámbitos compatibles y URL de documentación.

Escalakit_environment_url

La URL de entorno en la sección Configuración.

Escalakit_client_id

La identificación del cliente mencionada en la sección Configuración.

Scalekit_client_secret

La clave secreta que genera en Configuración → Credenciales de API.

Scalekit_resource_metadata_url

La URL expuesta por su servidor MCP para solicitudes de metadatos.

Ejemplo:

http://localhost:10000/.well-known/oauth-protected-resource/mcp

Scalekit_authorization_servers

La URL que apunta al identificador del servidor MCP emitido por ScaleKit.

Ejemplo:

https://<your-subdomain>.scalekit.dev/resources/res_***************

Puede encontrar el subdominio de los metadatos de recursos JSON

Scalekit_audience_name

La audiencia (AUD) reclama utilizados en los tokens de acceso para validar las solicitudes. Mira el Códigos completos aquí.

http://localhost:10000/mcp/

Scalekit_resouce_name

El nombre de recurso para su servidor MCP. En la mayoría de los casos, esto es lo mismo que Scalekit_audience_name. Mira el Códigos completos aquí.

Escalakit_resouce_docs_url

La URL donde se aloja la documentación de su servidor MCP.

Ejemplo:

http://localhost:10000/mcp/docs

Primero crearemos un archivo de configuración para cargar todas las variables de entorno que se utilizarán más adelante. Mira el Códigos completos aquí.

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_RESOUCE_NAME = os.environ.get('SCALEKIT_RESOUCE_NAME')
    SCALEKIT_RESOUCE_DOCS_URL = os.environ.get('SCALEKIT_RESOUCE_DOCS_URL')
    PORT = 10000
    

settings = Settings()

Este bloque de código obtiene datos de sentimientos de noticias en tiempo real para un ticker de stock dado utilizando la API Alpha Vantage. Recupera los tres primeros artículos recientes, resumiendo su título, resumen, fuente y tiempo de publicación para obtener información rápida. Mira el Códigos completos aquí.

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

# Create an MCP server
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:
    """Generic async caller to Alpha Vantage."""
    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:
    """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)

Este middleware actúa como una capa de autorización para su servidor MCP, asegurando que solo se procesen las solicitudes autenticadas. Usa el Cliente de kit de escala Para validar los tokens de acceso en cada solicitud entrante. Cuando entra una solicitud, el middleware primero verifica si la ruta es pública, como los puntos finales de metadatos en /.bien conocido/.

Si la solicitud no es para una ruta pública, busca un encabezado de autorización con un token portador válido. El token luego se valida utilizando SCALEKIT. Si falta el token, no es válido o expirado, el middleware responde inmediatamente con un 401 error no autorizado y un mensaje de error estructurado. Mira el Códigos completos aquí.

Si el token es válido, la solicitud se pasa a la siguiente capa de la aplicación. Además, el registro está integrado a lo largo del proceso para capturar eventos clave, lo que facilita la depuración y la auditoría de los flujos de autenticación.

Finalmente, este middleware se importará y se agregará al archivo del servidor para proteger todos los puntos finales seguros. Mira el Códigos completos aquí.

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

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

# Security scheme for Bearer token
security = HTTPBearer()

# Initialize ScaleKit client
scalekit_client = ScalekitClient(
    settings.SCALEKIT_ENVIRONMENT_URL,
    settings.SCALEKIT_CLIENT_ID,
    settings.SCALEKIT_CLIENT_SECRET
)

# Authentication middleware
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()
            
            # Parse JSON from bytes
            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)

Este script establece una aplicación FASTAPI integrada con un servidor MCP para el análisis de sentimientos de noticias. Comienza importando las bibliotecas necesarias, incluidas Fastapi, Cors Middleware y un middleware de autenticación personalizado. Mira el Códigos completos aquí.

El ciclo de vida de la aplicación se gestiona a través de un contexto combinado de vida útil utilizando un administrador de contexto asíncrono, asegurando que el finance_news_server.Session_Manager, que es esencialmente la lógica de sentimiento de stock que creamos, se ejecuta sin problemas durante el tiempo de ejecución de la aplicación. Cors Middleware está configurado para permitir solicitudes de origen cruzado, lo cual es útil durante el desarrollo, pero debe restringirse en entornos de producción.

Un nuevo punto final, /.well- conocido/oauth-protected-resource/mcpse agrega para servir metadatos para el descubrimiento de recursos protegidos de OAuth 2.1. Este punto final proporciona detalles importantes, como servidores de autorización compatibles, métodos de token de portador, nombre de recursos, URL de documentación y ámbitos compatibles, en este caso, MCP: Herramientas: Noticias: Leer.

El servidor MCP se crea utilizando la función Finance_News_Server.streamable_http_app () y se monta en la ruta de la raíz /, lo que hace que las funcionalidades MCP principales sean accesibles a través de la aplicación principal. La autenticación se aplica integrando el Authmiddlewarey el script asegura que este middleware se agregue correctamente al archivo del servidor.

Finalmente, la función Main () ejecuta la aplicación usando UVICORN, con el registro habilitado en el nivel de depuración, vinculando el servidor a localhost en el puerto configurado. Mira el Códigos completos aquí.

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

# Create a combined lifespan to manage the MCP session manager
@contextlib.asynccontextmanager
async def lifespan(app: FastAPI):
    async with finance_news_server.session_manager.run():
        yield

app = FastAPI(lifespan=lifespan)

# Add CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # In production, specify your actual origins
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
    allow_headers=["*"],
)

# MCP well-known endpoint
@app.get("/.well-known/oauth-protected-resource/mcp")
async def oauth_protected_resource_metadata():
    """
    OAuth 2.0 Protected Resource Metadata endpoint for MCP client discovery.
    Required by the MCP specification for authorization server discovery.
    """

    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"
        ],
    }

# Create and mount the MCP server with authentication
mcp_server = finance_news_server.streamable_http_app()
app.add_middleware(AuthMiddleware)
app.mount("/", mcp_server)

def main():
    """Main entry point for the MCP server."""
    uvicorn.run(app, host="localhost", port=settings.PORT, log_level="debug")

if __name__ == "__main__":
    main()

Para ejecutar el servidor, ejecute Python Server.py, que iniciará la aplicación en Localhost: 10000. Para probar la configuración, abra otro terminal y ejecute:

npx @modelcontextprotocol/inspector

Una vez que el inspector de MCP se ejecuta, ingrese http: // localhost: 10000/mcp Como la URL del servidor. Si intenta conectarse sin proporcionar credenciales válidas, encontrará el siguiente error:

Error de conexión: verifique si su servidor MCP se está ejecutando y si el apoderado El token está configurado correctamente.

Ahora, proporcione al token de portador utilizando la identificación secreta que generó en el kit de escala. Una vez ingresado, se autenticará con éxito y puede comenzar a hacer llamadas de herramientas.

Mira el Códigos completos aquí. No dude en ver nuestro Página de Github para tutoriales, códigos y cuadernos. Además, siéntete libre de seguirnos Gorjeo Y no olvides unirte a nuestro Subreddit de 100k+ ml y suscribirse a Nuestro boletín.

Soy un graduado de ingeniería civil (2022) de Jamia Millia Islamia, Nueva Delhi, y tengo un gran interés en la ciencia de datos, especialmente las redes neuronales y su aplicación en varias áreas.