Come configurare un server MCP con Okta e un reverse proxy in Python

In questo tutorial imparerai a configurare un server MCP (Model Context Protocol) remoto usando Streamable HTTP e un’autenticazione sicura con Okta, il tutto protetto da un reverse proxy in Python. MCP è uno standard open source che permette di collegare le applicazioni di intelligenza artificiale a sistemi esterni come database, strumenti, API e flussi di lavoro personalizzati. In questo modo agenti e modelli di IA possono accedere a informazioni chiave ed eseguire attività automatizzate in modo controllato e sicuro.

Il protocollo MCP può funzionare su diversi trasporti:

• In modalità stdio, la comunicazione avviene tramite input/output standard, ideale per l’integrazione locale.
• In modalità Streamable HTTP, il server MCP funziona come un processo indipendente accessibile via HTTP, consentendo connessioni remote.

In questo esempio pratico ci concentreremo sul trasporto Streamable HTTP. Configureremo un server MCP che espone uno strumento di ricerca di notizie (come esempio di integrazione), autenticando l’accesso con Okta e DPoP per la massima sicurezza. Implementeremo inoltre un reverse proxy per abilitare HTTPS e garantire che la comunicazione sia cifrata e protetta, anche se il server MCP gira internamente su HTTP.

Avviso: questo tutorial ha scopo didattico. Rivedi, adatta e metti in sicurezza la tua implementazione prima di usarla in produzione.

---

Generare una chiave privata DPoP per l’autenticazione con Okta

DPoP (Demonstration of Proof-of-Possession) è un meccanismo di sicurezza di OAuth che garantisce che i token possano essere usati solo dall’entità che li ha richiesti. Per prima cosa generiamo una chiave privata per firmare i JWT DPoP richiesti da Okta.

DPoP.py

from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import serialization

private_key = ec.generate_private_key(ec.SECP256R1())
private_pem = private_key.private_bytes(
    encoding=serialization.Encoding.PEM,
    format=serialization.PrivateFormat.PKCS8,
    encryption_algorithm=serialization.NoEncryption()
)
with open("dpop_private.pem", "wb") as f:
    f.write(private_pem)

Nota: questo script crea una chiave privata EC (curva P-256), la serializza in formato PEM e la salva nel file dpop_private.pem. Il client userà questo file per firmare i JWT DPoP durante l’autenticazione con Okta.

---

Costruire il server MCP con la verifica dei JWT di Okta

Il server MCP espone uno strumento per ottenere notizie ed è protetto da Okta tramite JWT, così solo i client autenticati possono accedere alle sue funzionalità.

mcp_server.py

import worldnewsapi
from worldnewsapi.rest import ApiException
from fastmcp import FastMCP
from fastmcp.server.auth.providers.jwt import JWTVerifier

# Configure JWTVerifier for Okta
auth = JWTVerifier(
    jwks_uri="https://zerogap.okta.com/oauth2/default/v1/keys",
    issuer="https://zerogap.okta.com/oauth2/default",
    audience="api://default"
)

# Initialize World News API client using API key
newsapi_key = "xxxxxx"
newsapi_config = worldnewsapi.Configuration(api_key={"apiKey": newsapi_key})
newsapi_client = worldnewsapi.NewsApi(worldnewsapi.ApiClient(newsapi_config))

# MCP server instance with Okta JWT authentication
mcp = FastMCP("News MCP Server", auth=auth)

# MCP tool for fetching latest news about a topic
@mcp.tool
def fetch_news(topic: str = "technology", max_results: int = 5):
    try:
        response = newsapi_client.search_news(
            text=topic,
            language="en",
            sort="publish-time",
            sort_direction="desc",
            number=max_results
        )
        news_list = []
        for article in response.news:
            news_item = {
                "title": getattr(article, "title", "No title available"),
                "url": getattr(article, "url", "No URL available"),
                "published": getattr(article, "publish_date", "No publish date available"),
            }
            if hasattr(article, "source"):
                news_item["source"] = article.source
            else:
                news_item["source"] = getattr(article, "author", "Unknown source")
            news_list.append(news_item)
        return news_list
    except ApiException as e:
        return {"error": f"News API error: {str(e)}"}
    except Exception as e:
        return {"error": f"Unexpected error: {str(e)}"}

if __name__ == "__main__":
    # Start an HTTP server on port 8089
    mcp.run(transport="http", host="0.0.0.0", port=8089)

Nota: questo server usa il verificatore JWT di Okta per autenticare le richieste ed espone uno strumento (fetch_news) che interroga la World News API e restituisce notizie. Il server gira su HTTP (porta 8089) e verrà poi protetto dal proxy.

---

Aggiungere un reverse proxy in Python per un accesso HTTPS sicuro

Per esporre il server MCP in modo sicuro useremo un semplice proxy con FastAPI. Così possiamo offrire HTTPS all’esterno mentre il server MCP continua a usare HTTP internamente.

reverseProxy.py

from fastapi import FastAPI, Request
from fastapi.responses import Response
import httpx

app = FastAPI()

TARGET_URL = "http://localhost:8089"  # Your target server on localhost:8089

@app.middleware("http")
async def reverse_proxy(request: Request, call_next):
    async with httpx.AsyncClient() as client:
        # Forward the request to the target server
        proxied_response = await client.request(
            method=request.method,
            url=TARGET_URL + request.url.path,
            headers=request.headers.raw,
            content=await request.body()
        )

        # Return the response from the target server
        return Response(
            content=proxied_response.content,
            status_code=proxied_response.status_code,
            headers=proxied_response.headers
        )

Nota: tutte le richieste al proxy vengono inoltrate al server MCP. Il proxy può girare con HTTPS/TLS usando Uvicorn, offrendo ai client un endpoint sicuro.

---

Creare il client MCP autenticato e l’assistente IA

Il client Python si autentica su Okta usando le client credentials e DPoP, poi si collega al server MCP tramite il proxy sicuro. Usa Azure OpenAI per l’interazione in linguaggio naturale e può cercare notizie tramite gli strumenti MCP.

MCPclient.py

import asyncio
import requests
import base64
import jwt
import time
import uuid
from cryptography.hazmat.primitives import serialization
from langchain.chat_models import AzureChatOpenAI
from langchain.agents import initialize_agent
from langchain.agents.agent_types import AgentType
from langchain_mcp_adapters.client import MultiServerMCPClient
from azure.identity import DefaultAzureCredential, get_bearer_token_provider
import httpx

# --- DPoP Helper Functions ---

def load_private_key():
    with open("dpop_private.pem", "rb") as f:
        return serialization.load_pem_private_key(f.read(), password=None)

def b64u(data):
    return base64.urlsafe_b64encode(data).rstrip(b'=').decode('ascii')

def make_dpop_proof(http_method, http_url, nonce=None):
    priv_key = load_private_key()
    pub_key = priv_key.public_key()
    numbers = pub_key.public_numbers()
    x = b64u(numbers.x.to_bytes(32, 'big'))
    y = b64u(numbers.y.to_bytes(32, 'big'))
    jwk = {
        "kty": "EC",
        "crv": "P-256",
        "x": x,
        "y": y
    }
    iat = int(time.time())
    jti = str(uuid.uuid4())
    payload = {
        "htu": http_url,
        "htm": http_method,
        "iat": iat,
        "jti": jti,
    }
    if nonce:
        payload["nonce"] = nonce
    headers = {
        "typ": "dpop+jwt",
        "alg": "ES256",
        "jwk": jwk
    }
    dpop_jwt = jwt.encode(
        payload,
        priv_key,
        algorithm="ES256",
        headers=headers
    )
    return dpop_jwt

# --- Okta Token Request with DPoP ---

def get_okta_access_token():
    OKTA_DOMAIN = "zerogap.okta.com"
    CLIENT_ID = "xxxxxx"
    CLIENT_SECRET = "xxxxxxx"
    TOKEN_URL = f"https://{OKTA_DOMAIN}/oauth2/default/v1/token"
    data = {
        "grant_type": "client_credentials",
        "scope": "MCPTest"
    }
    headers = {
        "Accept": "application/json",
        "Content-Type": "application/x-www-form-urlencoded",
        "DPoP": make_dpop_proof("POST", TOKEN_URL)
    }
    response = requests.post(
        TOKEN_URL,
        data=data,
        auth=(CLIENT_ID, CLIENT_SECRET),
        headers=headers
    )
    if response.status_code == 400 and "DPoP-Nonce" in response.headers:
        nonce = response.headers["DPoP-Nonce"]
        # Regenerate DPoP with nonce and try again
        headers["DPoP"] = make_dpop_proof("POST", TOKEN_URL, nonce=nonce)
        response = requests.post(
            TOKEN_URL,
            data=data,
            auth=(CLIENT_ID, CLIENT_SECRET),
            headers=headers
        )
    response.raise_for_status()
    return response.json()["access_token"]

# --- Main Async Client ---

async def main():
    okta_token = get_okta_access_token()

    mcp_connections = {
        "worldnews": {
            "url": 'https://localhost:8443/mcp',  # HTTPS endpoint
            "transport": "streamable_http",
            "headers": {
                "Authorization": f"Bearer {okta_token}"
            }        }
    }

    mcp_client = MultiServerMCPClient(connections=mcp_connections)
    tools = await mcp_client.get_tools()

    # Azure OpenAI LLM configuration
    azure_ad_token_provider = get_bearer_token_provider(
        DefaultAzureCredential(), "https://cognitiveservices.azure.com/.default"
    )
    endpoint = "https://zerogap.openai.azure.com/"
    deployment = "zerogap-gpt-4o"
    api_version = "2024-12-01-preview"
    llm = AzureChatOpenAI(
        azure_endpoint=endpoint,
        azure_ad_token_provider=azure_ad_token_provider,
        api_version=api_version,
        deployment_name=deployment,
        temperature=0
    )

    agent = initialize_agent(
        tools=tools,
        llm=llm,
        verbose=True,
        agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
        handle_parsing_errors=True,
    )

    print("Welcome to the AI Assistant! I can help with general questions and fetch news when needed.")
    print("Type 'exit' to quit.")

    while True:
        user_input = input("\nHow can I help you today? ")
        if user_input.lower() == 'exit':
            break
        response = await agent.arun(user_input)
        print(f"\nAssistant: {response}\n")

if __name__ == "__main__":
    asyncio.run(main())

Nota: questo client carica la chiave privata DPoP e crea JWT firmati per Okta, si autentica e gestisce le sfide del nonce, si collega al server MCP tramite il proxy sicuro e usa LangChain e Azure OpenAI per un assistente in grado di cercare notizie e rispondere alle domande.

---

Come mettere tutto in funzione

Passo dopo passo:

1. Genera la chiave DPoP
   • python DPoP.py
2. Avvia il server MCP
   • python mcp_server.py
3. Esegui il reverse proxy (con HTTPS)
   • uvicorn reverseProxy:app --host localhost --port 8443 --ssl-keyfile=path/to/key.pem --ssl-certfile=path/to/cert.pem
4. Avvia il client MCP
   • python MCPclient.py

---

Conclusione

Questo esempio mostra come distribuire un server MCP remoto usando il trasporto Streamable HTTP, ottenendo una comunicazione sicura e autenticata con Okta e DPoP e proteggendo l’accesso con un reverse proxy in Python. Anche se lo strumento di ricerca di notizie è solo un esempio, il vero valore di MCP sta nella sua capacità di standardizzare il collegamento tra agenti di IA e sistemi esterni, permettendo alle tue applicazioni di intelligenza artificiale di accedere ai dati, eseguire azioni e integrarsi con le infrastrutture aziendali in modo flessibile e sicuro.

Il protocollo MCP semplifica l’estensione dei tuoi agenti, permettendo di aggiungere nuovi strumenti, connettori e flussi di lavoro in base alle esigenze della tua organizzazione. Se cerchi un’architettura robusta e scalabile per potenziare le tue applicazioni di IA, MCP è lo standard ideale per collegare, orchestrare e proteggere l’interazione tra modelli, agenti e sistemi esterni.