Comment configurer un serveur MCP avec Okta et un reverse proxy en Python
Traduit de l'original en espagnol. Lire en espagnol
Dans ce tutoriel, vous apprendrez à configurer un serveur MCP (Model Context Protocol) distant en Streamable HTTP avec une authentification sécurisée via Okta, le tout protégé par un reverse proxy en Python. MCP est un standard open source qui permet de connecter des applications d’intelligence artificielle à des systèmes externes comme des bases de données, des outils, des API et des flux de travail personnalisés. Les agents et modèles d’IA peuvent ainsi accéder à des informations clés et exécuter des tâches automatisées de manière contrôlée et sécurisée.
Le protocole MCP peut fonctionner sur différents transports :
- En mode stdio, la communication passe par l’entrée/sortie standard, idéal pour une intégration locale.
- En mode Streamable HTTP, le serveur MCP fonctionne comme un processus indépendant accessible via HTTP, ce qui permet des connexions distantes.
Dans cet exemple pratique, nous nous concentrerons sur le transport Streamable HTTP. Nous configurerons un serveur MCP qui expose un outil de recherche d’actualités (à titre d’exemple d’intégration), en authentifiant l’accès avec Okta et DPoP pour une sécurité maximale. Nous mettrons également en place un reverse proxy pour activer HTTPS et garantir que la communication est chiffrée et protégée, même si le serveur MCP tourne en interne sur HTTP.
Avertissement : ce tutoriel a une visée pédagogique. Vérifiez, adaptez et sécurisez votre implémentation avant de l’utiliser en production.
Générer une clé privée DPoP pour l’authentification avec Okta
DPoP (Demonstration of Proof-of-Possession) est un mécanisme de sécurité OAuth qui garantit que les tokens ne peuvent être utilisés que par l’entité qui les a demandés. Nous générons d’abord une clé privée pour signer les JWT DPoP exigés par 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)Remarque : ce script crée une clé privée EC (courbe P-256), la sérialise au format PEM et l’enregistre dans le fichier dpop_private.pem. Le client utilisera ce fichier pour signer les JWT DPoP lors de l’authentification avec Okta.
Construire le serveur MCP avec vérification des JWT Okta
Le serveur MCP expose un outil de récupération d’actualités et il est protégé par Okta à l’aide de JWT, de sorte que seuls les clients authentifiés peuvent accéder à ses fonctionnalités.
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)Remarque : ce serveur utilise le vérificateur JWT d’Okta pour authentifier les requêtes et expose un outil (fetch_news) qui interroge la World News API et renvoie des actualités. Le serveur tourne en HTTP (port 8089) et sera ensuite protégé par le proxy.
Ajouter un reverse proxy en Python pour un accès HTTPS sécurisé
Pour exposer le serveur MCP de façon sécurisée, nous utiliserons un proxy simple avec FastAPI. Nous pouvons ainsi proposer HTTPS à l’extérieur tandis que le serveur MCP continue d’utiliser HTTP en interne.
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
)Remarque : toutes les requêtes adressées au proxy sont transmises au serveur MCP. Le proxy peut fonctionner en HTTPS/TLS avec Uvicorn, offrant ainsi un endpoint sécurisé aux clients.
Créer le client MCP authentifié et l’assistant IA
Le client Python s’authentifie auprès d’Okta avec des client credentials et DPoP, puis se connecte au serveur MCP via le proxy sécurisé. Il utilise Azure OpenAI pour l’interaction en langage naturel et peut rechercher des actualités grâce aux outils 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())Remarque : ce client charge la clé privée DPoP et crée des JWT signés pour Okta, s’authentifie et gère les défis de nonce, se connecte au serveur MCP via le proxy sécurisé, et utilise LangChain et Azure OpenAI pour un assistant capable de rechercher des actualités et de répondre aux questions.
Comment tout faire fonctionner
Pas à pas :
- Générez la clé DPoP
python DPoP.py
- Lancez le serveur MCP
python mcp_server.py
- Exécutez le reverse proxy (en HTTPS)
uvicorn reverseProxy:app --host localhost --port 8443 --ssl-keyfile=path/to/key.pem --ssl-certfile=path/to/cert.pem
- Lancez le client MCP
python MCPclient.py
Conclusion
Cet exemple montre comment déployer un serveur MCP distant avec le transport Streamable HTTP, en assurant une communication sécurisée et authentifiée grâce à Okta et DPoP, et en protégeant l’accès par un reverse proxy en Python. Même si l’outil de recherche d’actualités n’est qu’un exemple, la vraie valeur de MCP réside dans sa capacité à standardiser la connexion entre agents d’IA et systèmes externes, permettant à vos applications d’intelligence artificielle d’accéder aux données, d’exécuter des actions et de s’intégrer aux infrastructures d’entreprise de manière flexible et sécurisée.
Le protocole MCP facilite l’extension de vos agents en permettant d’ajouter de nouveaux outils, connecteurs et flux de travail selon les besoins de votre organisation. Si vous cherchez une architecture robuste et évolutive pour vos applications d’IA, MCP est le standard idéal pour connecter, orchestrer et protéger l’interaction entre modèles, agents et systèmes externes.
AI Platform Engineer & Full-Stack Developer
Building Enterprise Integrations & Automations
- Ce qu'est vraiment un LLM — et comment le mesurer avec Artificial Analysis - 12 septembre 2026
- AI SDLC : un recueil de bonnes pratiques pour développer avec l'IA - 19 juin 2026
- Agentic Patterns et les multiples visages du RAG - 25 mai 2026