PT

Como configurar um servidor MCP com Okta e um reverse proxy em Python

Traduzido do original em espanhol. Ler em espanhol

Neste tutorial, você vai aprender a configurar um servidor MCP (Model Context Protocol) remoto usando Streamable HTTP e autenticação segura com o Okta, tudo protegido por um reverse proxy em Python. O MCP é um padrão open-source que permite conectar aplicações de inteligência artificial a sistemas externos como bancos de dados, ferramentas, APIs e fluxos de trabalho personalizados. Assim, agentes e modelos de IA podem acessar informações importantes e executar tarefas automatizadas de forma controlada e segura.

O protocolo MCP pode funcionar sobre diferentes transportes:

  • No modo stdio, a comunicação é feita pela entrada/saída padrão, ideal para integração local.
  • No modo Streamable HTTP, o servidor MCP funciona como um processo independente acessível via HTTP, permitindo conexões remotas.

Neste exemplo prático, vamos nos concentrar no transporte Streamable HTTP. Vamos configurar um servidor MCP que expõe uma ferramenta de consulta de notícias (como exemplo de integração), autenticando o acesso com Okta e DPoP para máxima segurança. Além disso, vamos implementar um reverse proxy para habilitar HTTPS e garantir que a comunicação seja criptografada e protegida, mesmo que o servidor MCP rode internamente sobre HTTP.

Aviso: Este tutorial tem fins educativos. Revise, adapte e proteja sua implementação antes de usá-la em produção.

iStock AI Generator

Gerar uma chave privada DPoP para autenticação com o Okta

O DPoP (Demonstration of Proof-of-Possession) é um mecanismo de segurança do OAuth que garante que os tokens só sejam usados pela entidade que os solicitou. Primeiro, geramos uma chave privada para assinar os JWT DPoP exigidos pelo 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: Este script cria uma chave privada EC (curva P-256), serializa-a no formato PEM e a salva no arquivo dpop_private.pem. Esse arquivo será usado pelo cliente para assinar os JWT DPoP na autenticação com o Okta.

Construir o servidor MCP com verificação JWT do Okta

O servidor MCP expõe uma ferramenta para obter notícias e é protegido pelo Okta usando JWT, de modo que só clientes autenticados possam acessar suas 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 usa o verificador JWT do Okta para autenticar as requisições e expõe uma ferramenta (fetch_news) que consulta a World News API e devolve notícias. O servidor roda em HTTP (porta 8089) e depois será protegido pelo proxy.

Adicionar um reverse proxy em Python para acesso seguro via HTTPS

Para expor o servidor MCP de forma segura, usaremos um proxy simples com FastAPI. Assim, podemos oferecer HTTPS externamente enquanto o servidor MCP continua 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 as requisições ao proxy são redirecionadas ao servidor MCP. O proxy pode rodar com HTTPS/TLS usando o Uvicorn, oferecendo um endpoint seguro para os clientes.

Criar o cliente MCP autenticado e o assistente de IA

O cliente Python se autentica no Okta usando client credentials e DPoP e depois se conecta ao servidor MCP pelo proxy seguro. Ele usa o Azure OpenAI para interação em linguagem natural e pode consultar notícias por meio das ferramentas 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 carrega a chave privada DPoP e cria JWT assinados para o Okta, se autentica e trata os desafios de nonce, se conecta ao servidor MCP pelo proxy seguro e usa LangChain e Azure OpenAI para um assistente que pode consultar notícias e responder perguntas.

Como colocar tudo para funcionar

Passo a passo:

  1. Gere a chave DPoP
    • python DPoP.py
  2. Inicie o servidor MCP
    • python mcp_server.py
  3. Execute o reverse proxy (com HTTPS)
    • uvicorn reverseProxy:app --host localhost --port 8443 --ssl-keyfile=path/to/key.pem --ssl-certfile=path/to/cert.pem
  4. Inicie o cliente MCP
    • python MCPclient.py

Conclusão

Este exemplo mostra como implantar um servidor MCP remoto usando o transporte Streamable HTTP, com comunicação segura e autenticada via Okta e DPoP e acesso protegido por um reverse proxy em Python. Embora a ferramenta de consulta de notícias seja apenas um exemplo, o verdadeiro valor do MCP está na capacidade de padronizar a conexão entre agentes de IA e sistemas externos, permitindo que suas aplicações de inteligência artificial acessem dados, executem ações e se integrem a infraestruturas corporativas de forma flexível e segura.

O protocolo MCP facilita a expansão dos seus agentes, permitindo adicionar novas ferramentas, conectores e fluxos de trabalho conforme as necessidades da sua organização. Se você busca uma arquitetura robusta e escalável para turbinar suas aplicações de IA, o MCP é o padrão ideal para conectar, orquestrar e proteger a interação entre modelos, agentes e sistemas externos.

Maximiliano Díaz Doglia

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

Publicado em: IA