Source code for aidputils.agents.tools.mcp.mcp_auth

from typing import Optional, Dict, Set

from aidputils.agents.auth.strategy.auth_strategy_base import AuthStrategy
from aidputils.agents.auth.strategy.no_auth_strategy import NoAuthStrategy
from aidputils.agents.auth.strategy.bearer_strategy import BearerTokenAuthStrategy
from aidputils.agents.auth.strategy.oauth_strategy import OAuthAuthStrategy
from aidputils.agents.auth.strategy.oci_resource_principal_strategy import OCIResourcePrincipalAuthStrategy

# Centralized definition of supported auth types for MCP HTTP client headers
SUPPORTED_AUTH_TYPES: Set[str] = {"NO_AUTH", "BEARER_TOKEN", "OAUTH", "OCI_RESOURCE_PRINCIPAL"}




def build_headers_from_auth(auth: Optional[dict]) -> Dict[str, str]:
    """
    Build HTTP Authorization headers for MCP HTTP client based on auth configuration.
    Returns only auth headers (e.g., Authorization), not common headers like Accept.
    """
    auth = auth or {}
    auth_type = auth.get("authType", "NO_AUTH")

    if not isinstance(auth_type, str):
        raise ValueError("auth.authType must be a string when provided")

    norm_type = auth_type.strip().upper()

    if norm_type not in SUPPORTED_AUTH_TYPES:
        raise ValueError(
            f"Unsupported MCP authType '{auth_type}'. Supported types: {sorted(SUPPORTED_AUTH_TYPES)}"
        )

    strategies: Dict[str, type[AuthStrategy]] = {
        "NO_AUTH": NoAuthStrategy,
        "BEARER_TOKEN": BearerTokenAuthStrategy,
        "OAUTH": OAuthAuthStrategy,
        # Future
        # "OCI_RESOURCE_PRINCIPAL": OCIResourcePrincipalAuthStrategy,
    }
    strategy_cls = strategies.get(norm_type)
    if not strategy_cls:
        # Defensive fallback
        return {}

    strategy = strategy_cls(auth)
    return strategy.build_auth_headers()



__all__ = ["SUPPORTED_AUTH_TYPES", "build_headers_from_auth"]