ES

Cómo configurar un servidor MCP con Okta y un reverse proxy en Python

En este tutorial aprenderás a configurar un servidor MCP (Model Context Protocol) remoto utilizando Streamable HTTP y autenticación segura con Okta, todo ello protegido por un reverse proxy en Python. MCP es un estándar open-source que permite conectar aplicaciones de inteligencia artificial con sistemas externos como bases de datos, herramientas, APIs y flujos de trabajo personalizados. Así, agentes y modelos de IA pueden acceder a información clave y ejecutar tareas automatizadas de forma controlada y segura.

El protocolo MCP puede funcionar sobre distintos transportes:

  • En modo stdio, la comunicación se realiza por la entrada/salida estándar, ideal para integración local.
  • En modo Streamable HTTP, el servidor MCP opera como un proceso independiente accesible vía HTTP, permitiendo conexiones remotas.

En este ejemplo práctico, nos enfocaremos en el transporte Streamable HTTP. Configuraremos un servidor MCP que expone una herramienta de consulta de noticias (como ejemplo de integración), autenticando el acceso con Okta y DPoP para máxima seguridad. Además, implementaremos un reverse proxy para habilitar HTTPS y asegurarnos de que la comunicación sea cifrada y protegida, incluso si el servidor MCP corre internamente sobre HTTP.

Aviso: Este tutorial es con fines educativos. Revisa, adapta y asegura tu implementación antes de usarla en producción.

iStock AI Generator

Generar una clave privada DPoP para autenticación con Okta

DPoP (Demonstration of Proof-of-Possession) es un mecanismo de seguridad en OAuth que garantiza que los tokens sólo sean usados por la entidad que los solicitó. Primero, generamos una clave privada para firmar los JWT DPoP que Okta requiere.

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: Este script crea una clave privada EC (curva P-256), la serializa en formato PEM y la guarda en el archivo dpop_private.pem. Este archivo será utilizado por el cliente para firmar los JWT DPoP en la autenticación con Okta.

Construir el servidor MCP con verificación JWT de Okta

El servidor MCP expone una herramienta para obtener noticias y está protegido por Okta usando JWT, de modo que sólo los clientes autenticados puedan acceder a sus funcionalidades.

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: Este servidor utiliza el verificador JWT de Okta para autenticar las solicitudes y expone una herramienta (fetch_news) que consulta la World News API y devuelve noticias. El servidor corre en HTTP (puerto 8089) y luego será protegido por el proxy.

Añadir un reverse proxy en Python para acceso seguro por HTTPS

Para exponer el servidor MCP de forma segura, utilizaremos un proxy simple con FastAPI. Así, podemos ofrecer HTTPS externamente mientras el servidor MCP sigue usando 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: Todas las solicitudes al proxy se redirigen al servidor MCP. El proxy puede ejecutarse con HTTPS/TLS usando Uvicorn, proporcionando un endpoint seguro para los clientes.

Crear el cliente MCP autenticado y el asistente de IA

El cliente Python se autentica en Okta usando credenciales de cliente y DPoP, y luego se conecta al MCP server a través del proxy seguro. Utiliza Azure OpenAI para interacción en lenguaje natural y puede consultar noticias mediante las herramientas 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: Este cliente carga la clave privada DPoP y crea JWT firmados para Okta, se autentica y gestiona desafíos de nonce, se conecta al servidor MCP a través del proxy seguro, y usa LangChain y Azure OpenAI para un asistente que puede consultar noticias y responder preguntas.

Cómo ponerlo todo en marcha

Paso a paso:

  1. Genera la clave DPoP
    • python DPoP.py
  2. Inicia el servidor MCP
    • python mcp_server.py
  3. Ejecuta el reverse proxy (con HTTPS)
    • uvicorn reverseProxy:app --host localhost --port 8443 --ssl-keyfile=path/to/key.pem --ssl-certfile=path/to/cert.pem
  4. Inicia el cliente MCP
    • python MCPclient.py

Conclusión

Este ejemplo demuestra cómo desplegar un servidor MCP remoto utilizando el transporte Streamable HTTP, logrando una comunicación segura y autenticada con Okta y DPoP, y protegiendo el acceso mediante un reverse proxy en Python. Aunque la herramienta de consulta de noticias es solo un ejemplo, el verdadero valor de MCP reside en su capacidad para estandarizar la conexión entre agentes de IA y sistemas externos, permitiendo que tus aplicaciones de inteligencia artificial accedan a datos, ejecuten acciones y se integren con infraestructuras empresariales de manera flexible y segura.

El protocolo MCP facilita la expansión de tus agentes, permitiendo agregar nuevas herramientas, conectores y flujos de trabajo según las necesidades de tu organización. Si buscas una arquitectura robusta y escalable para potenciar tus aplicaciones de IA, MCP es el estándar ideal para conectar, orquestar y proteger la interacción entre modelos, agentes y sistemas externos.

Maximiliano Díaz Doglia

AI Platform Engineer & Full-Stack Developer
Building Enterprise Integrations & Automations

Publicado en: AI