How to Use MCP Servers with Okta Tokens and Enterprise APIs

January 15, 2026

Building AI-powered applications that integrate with enterprise systems requires secure authentication and seamless API connectivity. This guide walks you through using Model Context Protocol (MCP) servers with Okta authentication to access your existing enterprise APIs.

What is MCP?

The Model Context Protocol (MCP) is an open standard that enables AI assistants such as Claude and other MCP-compatible clients to securely connect with external tools, APIs, and data sources. Think of MCP as a universal adapter that allows AI models to interact with your enterprise systems while maintaining security and governance.

Why Okta + MCP?

Enterprise environments rely on identity providers like Okta for centralized authentication. By combining Okta tokens with MCP servers, you get:

Single Sign-On (SSO) integration with your existing identity infrastructure
Token-based authentication that leverages your organization's security policies
Audit trails through Okta's logging capabilities
Role-based access control inherited from your enterprise IAM

Architecture Overview

┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│   Claude /  │────▶│ MCP Server  │────▶│ Enterprise  │
│  MCP Client │     │  (Bridge)   │     │    API      │
└─────────────┘     └─────────────┘     └─────────────┘

How MCP Clients Pass Tokens to Servers

This is the critical question. The mechanism for passing authentication tokens depends on the transport type:

Transport Types and Token Passing

Transport	How Tokens Are Passed	Use Case
HTTP/SSE (Remote)	`Authorization: Bearer <token>` header	Remote MCP servers, web-based clients
STDIO (Local)	Environment variables or tool parameters	Claude Desktop, local CLI tools

Pattern 1: HTTP Transport with Bearer Token (Recommended for Remote Servers)

For HTTP-based MCP servers, the client passes the Okta token via the Authorization header. The server extracts it using middleware:

from fastmcp import FastMCP, Context
from fastmcp.server.dependencies import get_http_headers
from fastmcp.server.middleware import Middleware, MiddlewareContext
from fastmcp.exceptions import ToolError
import httpx

mcp = FastMCP("enterprise-api-server")

# ============== AUTHENTICATION MIDDLEWARE ==============
# This intercepts all requests and extracts the Bearer token from HTTP headers

class OktaAuthMiddleware(Middleware):
    """
    Middleware that extracts and validates Okta tokens from HTTP headers.
    The token is passed by the MCP client in the Authorization header
    and validated via the Okta OAuth 2.0 introspection endpoint.
    """
    
    async def on_call_tool(self, context: MiddlewareContext, call_next):
        """Called before each tool invocation."""
        headers = get_http_headers()
        auth_header = headers.get("authorization", "")
        
        if not auth_header.startswith("Bearer "):
            raise ToolError("Access denied: Missing or invalid Bearer token")
        
        token = auth_header.removeprefix("Bearer ").strip()
        
        # Validate token and extract user info
        user_info = await self.validate_and_get_user(token)
        if not user_info:
            raise ToolError("Access denied: Invalid or expired token")
        
        # Store user info in context for tools to access
        context.fastmcp_context.set_state("user_gpid", user_info["gpid"])
        context.fastmcp_context.set_state("okta_token", token)
        
        return await call_next(context)
    
    async def validate_and_get_user(self, token: str) -> dict | None:
        """Validate token with Okta and extract user claims."""
        try:
            async with httpx.AsyncClient() as client:
                response = await client.post(
                    f"{OKTA_DOMAIN}/oauth2/{AUTH_SERVER_ID}/v1/introspect",
                    headers={
                        "Authorization": f"Basic {ENCODED_CREDENTIALS}",
                        "Content-Type": "application/x-www-form-urlencoded"
                    },
                    data={"token": token, "token_type_hint": "access_token"}
                )
                result = response.json()

                if not result.get("active"):
                    return None

                # Custom attributes like GPID must be configured as custom claims in Okta
                return {
                    "gpid": result.get("gpid") or result.get("employeeNumber") or result.get("sub"),
                    "email": result.get("email"),
                    "name": result.get("name")
                }
        except Exception:
            return None

# Register the middleware
mcp = FastMCP("enterprise-api-server", middleware=[OktaAuthMiddleware()])

# ============== TOOLS ==============
# Tools access the token from context - NO token parameter needed!

@mcp.tool()
async def get_employee_information(ctx: Context) -> dict:
    """
    Retrieve employee information for the authenticated user.
    
    The Okta token is automatically extracted from the HTTP Authorization header
    by the middleware - the tool doesn't need to accept it as a parameter.
    """
    gpid = ctx.get_state("user_gpid")
    token = ctx.get_state("okta_token")
    
    return await call_enterprise_api("hr/employees", token, {"gpid": gpid})

@mcp.tool()
async def get_employee_schedule(
    ctx: Context,
    start_date: str, 
    end_date: str
) -> dict:
    """
    Retrieve employee schedule for a date range.
    
    Args:
        start_date: Start date (YYYY-MM-DD)
        end_date: End date (YYYY-MM-DD)
    
    Note: Authentication is handled via HTTP headers, not parameters.
    """
    gpid = ctx.get_state("user_gpid")
    token = ctx.get_state("okta_token")
    
    return await call_enterprise_api(
        "wfm/schedules",
        token,
        {"gpid": gpid, "start": start_date, "end": end_date}
    )

# Run as HTTP server
if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)

How the MCP client passes the token:

from fastmcp import Client
from fastmcp.client.auth import BearerAuth

# Client automatically includes token in Authorization header
async with Client(
    "https://your-mcp-server.com/mcp",
    auth=BearerAuth(token="eyJhbGciOiJSUzI1NiIs...")
) as client:
    # Token is passed via header - tools don't need token parameter
    result = await client.call_tool("get_employee_information", {})

Pattern 2: STDIO Transport with Token as Tool Parameter

For STDIO-based servers (commonly used with Claude Desktop), there is no HTTP layer. Tokens must be supplied via environment variables or tool parameters.

Warning: STDIO transport should only be used in trusted, single-user environments. Passing OAuth access tokens as tool parameters is not recommended for shared systems.

from fastmcp import FastMCP
import httpx

mcp = FastMCP("enterprise-api-server")

# ============== TOOLS ==============
# For STDIO transport, tokens must be passed as parameters

@mcp.tool()
async def get_employee_information(access_token: str) -> dict:
    """
    Retrieve employee information using Okta authentication.
    
    Args:
        access_token: Your Okta access token (obtain via `okta login` CLI or SSO)
    
    For STDIO-based MCP servers, the token must be passed as a parameter
    since there are no HTTP headers available.
    """
    # Validate token and extract GPID
    gpid = await validate_and_extract_gpid(access_token)
    return await call_enterprise_api("hr/employees", access_token, {"gpid": gpid})

@mcp.tool()
async def get_employee_schedule(
    access_token: str,
    start_date: str, 
    end_date: str
) -> dict:
    """
    Retrieve employee schedule for a date range.
    
    Args:
        access_token: Your Okta access token
        start_date: Start date (YYYY-MM-DD)
        end_date: End date (YYYY-MM-DD)
    """
    gpid = await validate_and_extract_gpid(access_token)
    return await call_enterprise_api(
        "wfm/schedules",
        access_token,
        {"gpid": gpid, "start": start_date, "end": end_date}
    )

# Run as STDIO server (for Claude Desktop)
if __name__ == "__main__":
    mcp.run(transport="stdio")

Pattern 3: STDIO with Environment Variable Token

For local development or when the same user always runs the server:

import os
from fastmcp import FastMCP

# Token loaded from environment at server startup
OKTA_TOKEN = os.environ.get("OKTA_ACCESS_TOKEN")

mcp = FastMCP("enterprise-api-server")

@mcp.tool()
async def get_employee_information() -> dict:
    """
    Retrieve employee information for the authenticated user.
    
    Uses the OKTA_ACCESS_TOKEN environment variable for authentication.
    Set this before starting the server.
    """
    if not OKTA_TOKEN:
        return {"error": "OKTA_ACCESS_TOKEN environment variable not set"}
    
    gpid = await validate_and_extract_gpid(OKTA_TOKEN)
    return await call_enterprise_api("hr/employees", OKTA_TOKEN, {"gpid": gpid})

Complete Implementation Guide

Step 1: Choose Your Transport and Token Pattern

Deployment	Transport	Token Mechanism
Remote server for multiple users	HTTP/SSE	Bearer token in header (Pattern 1)
Claude Desktop / local CLI	STDIO	Token as tool parameter (Pattern 2)
Single-user local development	STDIO	Environment variable (Pattern 3)

Step 2: Implement Token Validation

Always validate Okta tokens before making API calls:

import httpx
import base64

OKTA_DOMAIN = "https://your-org.okta.com"
SERVICE_CLIENT_ID = "your-service-client-id"
SERVICE_CLIENT_SECRET = "your-service-client-secret"

# Pre-compute encoded credentials
credentials = f"{SERVICE_CLIENT_ID}:{SERVICE_CLIENT_SECRET}"
ENCODED_CREDENTIALS = base64.b64encode(credentials.encode()).decode()

async def validate_and_extract_gpid(okta_token: str) -> str:
    """
    Validate the Okta token and extract the user's GPID.
    
    Args:
        okta_token: The Okta access token to validate
    
    Returns:
        The user's GPID extracted from token claims
    
    Raises:
        AuthenticationError: If token is invalid or expired
    """
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{OKTA_DOMAIN}/oauth2/v1/introspect",
            headers={
                "Authorization": f"Basic {ENCODED_CREDENTIALS}",
                "Content-Type": "application/x-www-form-urlencoded"
            },
            data={
                "token": okta_token,
                "token_type_hint": "access_token"
            }
        )
        response.raise_for_status()
        result = response.json()

        if not result.get("active"):
            raise AuthenticationError("Token is invalid or expired")

        gpid = result.get("gpid") or result.get("employeeNumber") or result.get("sub")
        if not gpid:
            raise AuthenticationError("Token does not contain user identifier")

        return gpid

Step 3: Implement Enterprise API Calls

ENTERPRISE_API_BASE = "https://api.yourcompany.com"

async def call_enterprise_api(
    endpoint: str,
    okta_token: str,
    params: dict = None
) -> dict:
    """
    Make authenticated calls to enterprise APIs.
    
    Args:
        endpoint: The API endpoint to call
        okta_token: Valid Okta access token
        params: Optional query parameters
    
    Returns:
        API response as a dictionary
    """
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{ENTERPRISE_API_BASE}/{endpoint}",
            headers={
                "Authorization": f"Bearer {okta_token}",
                "Content-Type": "application/json"
            },
            params=params
        )
        response.raise_for_status()
        return response.json()

Step 4: Add Resources for Static Data

Resources are for read-only, static data that doesn't require per-request authentication:

# ============== RESOURCES ==============
# Resources expose static/semi-static data via URIs
# They do NOT accept authentication parameters

@mcp.resource("config://app/settings")
def get_app_settings() -> dict:
    """
    Application configuration - no auth required.
    
    Resources are for static data. For user-specific data
    that requires authentication, use Tools instead.
    """
    return {
        "app_name": "Enterprise HR System",
        "version": "2.1.0",
        "api_base": ENTERPRISE_API_BASE,
        "supported_date_formats": ["YYYY-MM-DD", "MM/DD/YYYY"]
    }

@mcp.resource("docs://api/time-off-policies")
def get_time_off_policies() -> str:
    """Time-off policy documentation."""
    return """
    Time-Off Request Policies
    =========================
    1. Requests must be submitted 2 weeks in advance
    2. Maximum consecutive days: 10 without manager approval
    3. Blackout periods: Dec 28-31, March 1-5
    """

@mcp.resource("schema://employee/{gpid}")
def get_employee_schema(gpid: str) -> dict:
    """
    Schema information for employee data.
    
    This is a resource TEMPLATE - the {gpid} comes from the URI.
    It returns metadata/schema, not actual employee data.
    """
    return {
        "gpid": gpid,
        "schema_version": "1.0",
        "fields": ["name", "department", "manager", "location"]
    }

Tools vs Resources: When to Use Each

Use Case	Use Tool or Resource?	Why
Get employee data (requires auth)	Tool	Needs Okta token
Get user's schedule (requires auth)	Tool	Needs Okta token
Submit time-off request	Tool	Side effect + auth
App configuration	Resource	Static, no auth
API documentation	Resource	Static, no auth
Data schemas	Resource	Static, no auth

Key Rule: If it requires authentication or has side effects → Tool. If it's static read-only data → Resource.

How Users Obtain Their Okta Token

Before using the MCP server, users need to obtain their Okta token.

Important: Users must obtain OAuth 2.0 access tokens, not Okta API tokens. Supported OAuth flows include:

Authorization Code + PKCE (user-facing applications)

Client Credentials (service-to-service)

Method 1: Programmatic (PKCE Flow)

$ ./get-my-okta-token.sh
Your Okta token: eyJhbGciOiJSUzI1NiIs...

Method 2: Company SSO Portal

Log into your company portal
Navigate to Developer Tools > OAuth Tokens
Copy your access token

Best Practices

Security

Never hardcode tokens — Use environment variables or secure vaults
Validate on every request — Don't cache token validation results
Use short-lived tokens — Configure appropriate expiration in Okta
Log access attempts — Maintain audit trails

Error Handling

class OktaAuthError(Exception):
    """Raised when Okta authentication fails."""
    pass

class EnterpriseAPIError(Exception):
    """Raised when the enterprise API returns an error."""
    pass

async def safe_api_call(okta_token: str, endpoint: str, params: dict):
    try:
        gpid = await validate_and_extract_gpid(okta_token)
        return await call_enterprise_api(endpoint, okta_token, params)
    
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 401:
            raise OktaAuthError("Token rejected by API")
        elif e.response.status_code == 404:
            raise EnterpriseAPIError("Resource not found")
        else:
            raise EnterpriseAPIError(f"API error: {e.response.status_code}")

Troubleshooting

Token Issues

Problem	Cause	Solution
"Invalid token"	Token expired	Get a new token
"Token rejected"	Wrong audience	Check Okta app config
"Missing Authorization header"	Client not sending token	Configure client auth

Transport Issues

Problem	Cause	Solution
"Cannot read headers"	Using STDIO transport	Pass token as parameter
"Connection refused"	Server not running	Check server is up
"SSL error"	Certificate issues	Update CA certificates

Conclusion

The key to MCP + Okta integration is understanding how tokens are passed:

HTTP transport: Token in Authorization: Bearer header, extracted via middleware
STDIO transport: Token as tool parameter or environment variable

Choose the pattern that matches your deployment:

Remote multi-user server → HTTP with middleware
Claude Desktop → STDIO with token parameter
Local single-user dev → STDIO with environment variable

For more information, see the FastMCP documentation, MCP Specification, and Okta Developer Documentation.