Documentation Index
Fetch the complete documentation index at: https://gofastmcp.com/llms.txt
Use this file to discover all available pages before exploring further.
Meet Prefect Horizon, the enterprise MCP gateway built by the team behind FastMCP
Documentation Index
Fetch the complete documentation index at: https://gofastmcp.com/llms.txt
Use this file to discover all available pages before exploring further.
fastmcp.server.auth.providers.ocifrom fastmcp import FastMCP
from fastmcp.server.auth.providers.oci import OCIProvider
from fastmcp.server.dependencies import get_access_token
from fastmcp.utilities.logging import get_logger
import os
import oci
from oci.auth.signers import TokenExchangeSigner
logger = get_logger(__name__)
# Load configuration from environment
config_url = os.environ.get("OCI_CONFIG_URL") # OCI IAM Domain OIDC discovery URL
client_id = os.environ.get("OCI_CLIENT_ID") # Client ID configured for the OCI IAM Domain Integrated Application
client_secret = os.environ.get("OCI_CLIENT_SECRET") # Client secret configured for the OCI IAM Domain Integrated Application
iam_guid = os.environ.get("OCI_IAM_GUID") # IAM GUID configured for the OCI IAM Domain
# Simple OCI OIDC protection
auth = OCIProvider(
config_url=config_url, # config URL is the OCI IAM Domain OIDC discovery URL
client_id=client_id, # This is same as the client ID configured for the OCI IAM Domain Integrated Application
client_secret=client_secret, # This is same as the client secret configured for the OCI IAM Domain Integrated Application
required_scopes=["openid", "profile", "email"],
redirect_path="/auth/callback",
base_url="http://localhost:8000",
)
# NOTE: For production use, replace this with a thread-safe cache implementation
# such as threading.Lock-protected dict or a proper caching library
_global_token_cache = {} # In memory cache for OCI session token signer
def get_oci_signer() -> TokenExchangeSigner:
authntoken = get_access_token()
tokenID = authntoken.claims.get("jti")
token = authntoken.token
# Check if the signer exists for the token ID in memory cache
cached_signer = _global_token_cache.get(tokenID)
logger.debug(f"Global cached signer: {cached_signer}")
if cached_signer:
logger.debug(f"Using globally cached signer for token ID: {tokenID}")
return cached_signer
# If the signer is not yet created for the token then create new OCI signer object
logger.debug(f"Creating new signer for token ID: {tokenID}")
signer = TokenExchangeSigner(
jwt_or_func=token,
oci_domain_id=iam_guid.split(".")[0] if iam_guid else None, # This is same as IAM GUID configured for the OCI IAM Domain
client_id=client_id, # This is same as the client ID configured for the OCI IAM Domain Integrated Application
client_secret=client_secret, # This is same as the client secret configured for the OCI IAM Domain Integrated Application
)
logger.debug(f"Signer {signer} created for token ID: {tokenID}")
#Cache the signer object in memory cache
_global_token_cache[tokenID] = signer
logger.debug(f"Signer cached for token ID: {tokenID}")
return signer
mcp = FastMCP("My Protected Server", auth=auth)
OCIProvider