Конфигурационный файл (config.py)

Мы создадим конфигурационный файл для загрузки всех переменных окружения, которые будут использоваться позже.

import os
from dotenv import load_dotenv

load_dotenv()

class Settings():
ALPHAVANTAGEAPIKEY = os.environ.get(‘ALPHAVANTAGEAPIKEY’)
METADATAJSONRESPONSE = os.environ.get(‘METADATAJSONRESPONSE’)
SCALEKITENVIRONMENTURL = os.environ.get(‘SCALEKITENVIRONMENTURL’)
SCALEKITCLIENTID = os.environ.get(‘SCALEKITCLIENTID’)
SCALEKITCLIENTSECRET = os.environ.get(‘SCALEKITCLIENTSECRET’)
SCALEKITRESOURCEMETADATAURL = os.environ.get(‘SCALEKITRESOURCEMETADATAURL’)
SCALEKITAUTHORIZATIONSERVERS = os.environ.get(‘SCALEKITAUTHORIZATIONSERVERS’)
SCALEKITAUDIENCENAME = os.environ.get(‘SCALEKITAUDIENCENAME’)
SCALEKITRESOUCENAME = os.environ.get(‘SCALEKITRESOUCENAME’)
SCALEKITRESOUCEDOCSURL = os.environ.get(‘SCALEKITRESOUCEDOCSURL’)
PORT = 10000

settings = Settings()

Логика получения настроений на фондовом рынке (finance.py)

Этот блок кода получает данные о настроениях на фондовом рынке для заданного тикера акций с помощью Alpha Vantage API. Он извлекает три последние статьи, суммируя их заголовок, краткое содержание, источник и время публикации для быстрого анализа.

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 callalphavantage(endpoint: str, params: dict[str, Any]) -> dict[str, Any] | None:
“””Generic async caller to Alpha Vantage.”””
params[“apikey”] = settings.ALPHAVANTAGEAPI_KEY
params[“function”] = endpoint
async with httpx.AsyncClient() as client:
try:
response = await client.get(BASE_URL, params=params, timeout=30.0)
response.raiseforstatus()
return response.json()
except Exception:
return None

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

Args:
ticker: Stock ticker symbol (e.g., MSFT, AAPL)
“””
data = await callalphavantage(“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, обеспечивая обработку только аутентифицированных запросов. Оно использует клиент ScaleKit для проверки токенов доступа на каждом входящем запросе.

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.SCALEKITENVIRONMENTURL,
settings.SCALEKITCLIENTID,
settings.SCALEKITCLIENTSECRET
)

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 authheader or not authheader.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:
requestdata = json.loads(requestbody.decode(‘utf-8’))
except (json.JSONDecodeError, UnicodeDecodeError):
request_data = {}

try:
scalekitclient.validateaccess_token(token)

except Exception as e:
raise HTTPException(status_code=401, detail=”Token validation failed”)

except HTTPException as e:
return JSONResponse(
statuscode=e.statuscode,
content={“error”: “unauthorized” if e.statuscode == 401 else “forbidden”, “errordescription”: e.detail},
headers={
“WWW-Authenticate”: f’Bearer realm=”OAuth”, resourcemetadata=”{settings.SCALEKITRESOURCEMETADATAURL}”‘
}
)

return await call_next(request)

Сервер MCP (server.py)

Этот скрипт настраивает приложение FastAPI, интегрированное с сервером MCP для анализа настроений на фондовом рынке.

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 financenewsserver

Create a combined lifespan to manage the MCP session manager

@contextlib.asynccontextmanager
async def lifespan(app: FastAPI):
async with financenewsserver.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 oauthprotectedresource_metadata():
“””
OAuth 2.0 Protected Resource Metadata endpoint for MCP client discovery.
Required by the MCP specification for authorization server discovery.
“””

return {
“authorizationservers”: [settings.SCALEKITAUTHORIZATION_SERVERS],
“bearermethodssupported”: [“header”],
“resource”: settings.SCALEKITRESOURCENAME,
“resourcedocumentation”: settings.SCALEKITRESOURCEDOCSURL,
“scopes_supported”: [
“mcp:tools:news:read”
],
}

Create and mount the MCP server with authentication

mcpserver = financenewsserver.streamablehttp_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()

Запуск сервера

Чтобы запустить сервер, выполните python server.py, который запустит приложение на localhost:10000. Чтобы протестировать настройку, откройте другой терминал и запустите:

npx @modelcontextprotocol/inspector

После запуска MCP Inspector откройте http://localhost:10000/mcp в качестве URL-адреса сервера. Если вы попытаетесь подключиться без предоставления действительных учётных данных, вы столкнётесь с ошибкой подключения.

1. Какие шаги необходимо выполнить для настройки Scalekit для работы с серверами MCP?

Ответ: для начала работы с Scalekit необходимо создать учётную запись на сайте scalekit.com, активировать Full-Stack Auth, настроить разрешения, добавив необходимое разрешение (например, `news:read` для использования Alpha Vantage для получения настроений на фондовом рынке), и добавить свой сервер MCP, указав его имя и уникальный идентификатор ресурса.

2. Какие инструменты и библиотеки используются для реализации OAuth 2.1 для серверов MCP в этом руководстве?

Ответ: в этом руководстве используются Scalekit для упрощения настройки OAuth, Node.js для запуска MCP Inspector, Python с библиотеками `fastapi`, `fastmcp`, `mcp` и `scalekit-sdk-python` для настройки зависимостей и работы с API.

3. Как получить API-ключ для Alpha Vantage и где его использовать в этом проекте?

Ответ: чтобы получить API-ключ для Alpha Vantage, необходимо посетить платформу Alpha Vantage, ввести свой email и необходимые данные, и получить ключ. Этот ключ используется в конфигурационном файле (config.py) для аутентификации запросов к Alpha Vantage API.

4. Какие переменные окружения необходимо настроить для работы с Scalekit и Alpha Vantage в этом проекте?

Ответ: для работы с Scalekit и Alpha Vantage необходимо настроить следующие переменные окружения: `ALPHAVANTAGEAPIKEY`, `SCALEKITENVIRONMENTURL`, `SCALEKITCLIENTID`, `SCALEKITCLIENTSECRET`, `SCALEKITRESOURCEMETADATAURL`, `SCALEKITAUTHORIZATIONSERVERS`, `SCALEKITAUDIENCENAME`, `SCALEKITRESOUCENAME`, `SCALEKITRESOUCEDOCS_URL`. Эти переменные используются в конфигурационном файле для загрузки настроек.

5. Как работает промежуточное ПО для авторизации в этом проекте и какие функции оно выполняет?

Ответ: промежуточное ПО для авторизации действует как авторизационный слой для сервера MCP, обеспечивая обработку только аутентифицированных запросов. Оно использует клиент ScaleKit для проверки токенов доступа на каждом входящем запросе. Если токен недействителен, возвращается ошибка HTTP 401.

Источник