EN

How to set up an MCP server with Okta and a reverse proxy in Python

Translated from the Spanish original. Read in Spanish

In this tutorial you’ll learn how to set up a remote MCP (Model Context Protocol) server using Streamable HTTP and secure authentication with Okta, all protected by a reverse proxy in Python. MCP is an open-source standard for connecting AI applications to external systems such as databases, tools, APIs and custom workflows. That way, AI agents and models can access key information and run automated tasks in a controlled and secure way.

The MCP protocol can run over different transports:

  • In stdio mode, communication happens over standard input/output — ideal for local integration.
  • In Streamable HTTP mode, the MCP server runs as a separate process reachable over HTTP, allowing remote connections.

In this hands-on example, we’ll focus on the Streamable HTTP transport. We’ll set up an MCP server that exposes a news lookup tool (as an integration example), authenticating access with Okta and DPoP for maximum security. We’ll also add a reverse proxy to enable HTTPS and make sure communication is encrypted and protected, even if the MCP server runs internally over HTTP.

Disclaimer: This tutorial is for educational purposes. Review, adapt and secure your implementation before using it in production.

iStock AI Generator

Generate a DPoP private key for authentication with Okta

DPoP (Demonstration of Proof-of-Possession) is an OAuth security mechanism that ensures tokens can only be used by the party that requested them. First, we generate a private key to sign the DPoP JWTs that Okta requires.

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)

Note: This script creates an EC private key (P-256 curve), serialises it in PEM format and saves it to the dpop_private.pem file. The client will use this file to sign the DPoP JWTs when authenticating with Okta.

Build the MCP server with Okta JWT verification

The MCP server exposes a tool for fetching news and is protected by Okta using JWT, so only authenticated clients can use its features.

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)

Note: This server uses Okta’s JWT verifier to authenticate requests and exposes a tool (fetch_news) that queries the World News API and returns news. The server runs over HTTP (port 8089) and will then be protected by the proxy.

Add a reverse proxy in Python for secure HTTPS access

To expose the MCP server securely, we’ll use a simple proxy built with FastAPI. That way we can offer HTTPS externally while the MCP server keeps using HTTP internally.

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
        )

Note: All requests to the proxy are forwarded to the MCP server. The proxy can run with HTTPS/TLS using Uvicorn, giving clients a secure endpoint.

Create the authenticated MCP client and the AI assistant

The Python client authenticates with Okta using client credentials and DPoP, then connects to the MCP server through the secure proxy. It uses Azure OpenAI for natural-language interaction and can look up news through the MCP tools.

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())

Note: This client loads the DPoP private key and creates signed JWTs for Okta, authenticates and handles nonce challenges, connects to the MCP server through the secure proxy, and uses LangChain and Azure OpenAI for an assistant that can look up news and answer questions.

How to get it all running

Step by step:

  1. Generate the DPoP key
    • python DPoP.py
  2. Start the MCP server
    • python mcp_server.py
  3. Run the reverse proxy (with HTTPS)
    • uvicorn reverseProxy:app --host localhost --port 8443 --ssl-keyfile=path/to/key.pem --ssl-certfile=path/to/cert.pem
  4. Start the MCP client
    • python MCPclient.py

Conclusion

This example shows how to deploy a remote MCP server using the Streamable HTTP transport, with secure, authenticated communication via Okta and DPoP, and access protected by a reverse proxy in Python. Although the news lookup tool is just an example, MCP’s real value lies in standardising the connection between AI agents and external systems, letting your AI applications access data, take actions and integrate with enterprise infrastructure flexibly and securely.

The MCP protocol makes it easy to extend your agents, adding new tools, connectors and workflows as your organisation needs them. If you’re looking for a robust, scalable architecture for your AI applications, MCP is the ideal standard for connecting, orchestrating and protecting the interaction between models, agents and external systems.

Maximiliano Díaz Doglia

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

Published in: AI