Authentication

Base Auth

agora_workbench.code_execution.auth.base

Abstract authentication interfaces for MCP code execution servers.

This module defines the pluggable authentication contract that decouples the server from any specific identity provider (Azure Entra ID, Auth0, Keycloak, etc.).

Three interfaces form the abstraction:

• TokenValidator: Validates bearer tokens and returns claims.
• IdentityExtractor: Derives a unique user identity string from claims.
• CredentialProvider: Provides credentials for downstream resource access.

Implementations for Azure Entra ID live in auth.entra. A no-op implementation suitable for local development lives in auth.noop.

AccessToken

Bases: NamedTuple

A provider-agnostic access token.

Structurally compatible with azure.core.credentials.AccessToken.

TokenValidationError(message, status_code=401)

Bases: Exception

Raised when token validation fails.

Source code in src/agora_workbench/code_execution/auth/base.py

def __init__(self, message: str, status_code: int = 401):
    self.status_code = status_code
    super().__init__(message)

TokenValidator

Bases: ABC

Validates bearer tokens from incoming requests.

Implementations handle signature verification, expiry checks, audience/issuer validation, and any provider-specific logic.

Implementations do not need to publish OAuth discovery metadata. Set AuthConfig.protected_resource_metadata instead — it is provider-agnostic and part of the public contract.

.. deprecated:: Servers also probe validators for private _client_id / _tenant_id attributes to compose Entra-shaped metadata. That fallback is retained for validators written against the old convention; new implementations should use AuthConfig.protected_resource_metadata.

validate(token, *, request_path='/mcp', request_method='POST') abstractmethod async

Validate a bearer token and return decoded claims.

Parameters:

Name	Type	Description	Default
token	str	The raw bearer token string.	required
request_path	str	Optional hint for validators that support path-based authorization. Not used by standard JWT validation.	'/mcp'
request_method	str	Optional hint for validators that support method-based authorization. Not used by standard JWT validation.	'POST'

Returns:

Type	Description
dict	A dictionary of validated claims (e.g. {"oid": "...", "tid": "...", "name": "..."}).

Raises:

Type	Description
TokenValidationError	If the token is invalid, expired, or fails any validation check.

Source code in src/agora_workbench/code_execution/auth/base.py

@abstractmethod
async def validate(self, token: str, *, request_path: str = "/mcp", request_method: str = "POST") -> dict:
    """
    Validate a bearer token and return decoded claims.

    Args:
        token: The raw bearer token string.
        request_path: Optional hint for validators that support path-based
            authorization. Not used by standard JWT validation.
        request_method: Optional hint for validators that support
            method-based authorization. Not used by standard JWT validation.

    Returns:
        A dictionary of validated claims (e.g. ``{"oid": "...", "tid": "...", "name": "..."}``).

    Raises:
        TokenValidationError: If the token is invalid, expired, or
            fails any validation check.
    """
    ...

IdentityExtractor

Bases: ABC

Extracts a unique user identity string from validated token claims.

The identity is used for session ownership and access control. Different providers encode identity differently (Azure uses oid+tid, generic OIDC uses sub, etc.).

extract(claims) abstractmethod

Derive a unique user identity from token claims.

Parameters:

Name	Type	Description	Default
claims	dict	Validated token claims dictionary.	required

Returns:

Type	Description
Optional[str]	A unique string identifying the user (e.g. "oid@tid"),
Optional[str]	or None if required claims are missing.

Source code in src/agora_workbench/code_execution/auth/base.py

@abstractmethod
def extract(self, claims: dict) -> Optional[str]:
    """
    Derive a unique user identity from token claims.

    Args:
        claims: Validated token claims dictionary.

    Returns:
        A unique string identifying the user (e.g. ``"oid@tid"``),
        or ``None`` if required claims are missing.
    """
    ...

CredentialProvider

Bases: ABC

Provides credentials for accessing downstream resources on behalf of the authenticated user or the server's own identity.

Implementations may use managed identity or provide static/no-op credentials depending on deployment context.

get_token(scope) abstractmethod async

Obtain an access token for the given resource scope.

Parameters:

Name	Type	Description	Default
scope	str	The target resource scope (e.g. "https://storage.azure.com/.default").	required

Returns:

Type	Description
AccessToken	An AccessToken with the token string and expiration.

Raises:

Type	Description
CredentialError	If token acquisition fails.

Source code in src/agora_workbench/code_execution/auth/base.py

@abstractmethod
async def get_token(self, scope: str) -> AccessToken:
    """
    Obtain an access token for the given resource scope.

    Args:
        scope: The target resource scope
               (e.g. ``"https://storage.azure.com/.default"``).

    Returns:
        An AccessToken with the token string and expiration.

    Raises:
        CredentialError: If token acquisition fails.
    """
    ...

close() async

Release any held resources.

Source code in src/agora_workbench/code_execution/auth/base.py

async def close(self) -> None:
    """Release any held resources."""

CredentialError(message, scope, original_error=None)

Bases: Exception

Raised when credential acquisition fails.

Source code in src/agora_workbench/code_execution/auth/base.py

def __init__(self, message: str, scope: str, original_error: Optional[Exception] = None):
    self.scope = scope
    self.original_error = original_error
    super().__init__(message)

AuthConfig(token_validator, identity_extractor, credential_provider_factory=None, www_authenticate_value='', require_authorization_header=True, protected_resource_metadata=None) dataclass

Composite auth configuration injected into CodeExecutionServer.

Bundles the three auth interfaces so servers can be constructed with a single auth_config parameter instead of scattered provider args.

credential_provider_factory = None class-attribute instance-attribute

Optional callable: (user_token: str) -> CredentialProvider.

Called per-session to create a credential provider scoped to the authenticated user's token. If None, downstream resource access is not available (suitable for servers that don't call external APIs).

www_authenticate_value = '' class-attribute instance-attribute

Value for the WWW-Authenticate response header on 401.

require_authorization_header = True class-attribute instance-attribute

Whether protected endpoints require an Authorization header.

protected_resource_metadata = None class-attribute instance-attribute

RFC 9728 document served verbatim at /.well-known/oauth-protected-resource.

Lets any identity provider describe itself without the server needing provider-specific knowledge. create_entra_auth_config() populates this with an Entra-shaped document; custom validators should set it explicitly.

When set, it must be a non-empty mapping containing at least resource, the only field RFC 9728 marks REQUIRED. This is validated on construction, so the endpoint can serve any non-None value as-is rather than silently returning a document that discovery clients cannot use.

When None, servers fall back to composing an Entra document from their entra_client_id / entra_tenant_id attributes.

Entra ID

agora_workbench.code_execution.auth.entra

Azure Entra ID authentication implementations.

Provides concrete implementations of the auth interfaces (TokenValidator, IdentityExtractor, CredentialProvider) for Microsoft Entra ID (Azure AD).

Uses PyJWT for JWT validation with JWKS key fetching from the Microsoft identity platform's OpenID configuration endpoint.

EntraTokenValidator(client_id, tenant_id)

Bases: TokenValidator

Validates JWTs issued by Microsoft Entra ID using PyJWT.

Fetches signing keys from the Microsoft identity platform JWKS endpoint, verifies signature (RS256), and validates standard claims (exp, aud, iss).

Note: This implementation is single-tenant only. The JWKS URL and issuer validation use the specific tenant ID. For multi-tenant support, use the /common/ JWKS endpoint and validate the tid claim manually.

Parameters:

Name	Type	Description	Default
client_id	str	The Entra ID application (client) ID — used as audience.	required
tenant_id	str	The Entra ID tenant ID.	required

Source code in src/agora_workbench/code_execution/auth/entra.py

def __init__(self, client_id: str, tenant_id: str):
    """
    Args:
        client_id: The Entra ID application (client) ID — used as audience.
        tenant_id: The Entra ID tenant ID.
    """
    self._client_id = client_id
    self._tenant_id = tenant_id
    jwks_url = _ENTRA_JWKS_URL_TEMPLATE.format(tenant_id=tenant_id)
    self._jwks_client = PyJWKClient(jwks_url, cache_keys=True)
    self._valid_audiences = [client_id, f"api://{client_id}"]
    self._valid_issuers = [t.format(tenant_id=tenant_id) for t in _ENTRA_ISSUER_TEMPLATES]

EntraIdentityExtractor

Bases: IdentityExtractor

Extracts user identity from Entra ID JWT claims as {oid}@{tid}.

This composite format ensures session isolation across both users and tenants. Falls back to sub if oid is absent.

EntraCredentialProvider(client_id=None)

Bases: CredentialProvider

Provides Azure credentials for downstream resource access via managed identity.

Uses Azure ManagedIdentityCredential (user-assigned if AZURE_CLIENT_ID is set, otherwise system-assigned) to obtain tokens for downstream resources such as Azure Storage and Azure AI Search.

Parameters:

Name	Type	Description	Default
client_id	Optional[str]	User-assigned managed identity client ID. Falls back to AZURE_CLIENT_ID env var. If None/unset, system-assigned managed identity is used.	None

Source code in src/agora_workbench/code_execution/auth/entra.py

def __init__(self, client_id: Optional[str] = None):
    """
    Args:
        client_id: User-assigned managed identity client ID.
                  Falls back to AZURE_CLIENT_ID env var.
                  If None/unset, system-assigned managed identity is used.
    """
    from azure.identity.aio import ManagedIdentityCredential

    resolved_client_id = client_id or (os.getenv("AZURE_CLIENT_ID") or "").strip() or None
    self._credential = ManagedIdentityCredential(client_id=resolved_client_id)
    if resolved_client_id:
        LOGGER.info(f"EntraCredentialProvider: using user-assigned MI (client_id={resolved_client_id[:8]}...)")
    else:
        LOGGER.info("EntraCredentialProvider: using system-assigned managed identity")

CredentialProviderTokenCredential(credential_provider)

Adapter exposing auth.CredentialProvider via Azure token-credential interface.

Source code in src/agora_workbench/code_execution/auth/entra.py

def __init__(self, credential_provider: CredentialProvider):
    self._credential_provider = credential_provider

create_entra_auth_config(client_id=None, tenant_id=None)

Create an AuthConfig using Azure Entra ID for all three auth concerns.

This is the default configuration for production and local-dev-with-Azure deployments. Reads from environment variables if arguments are not provided.

Parameters:

Name	Type	Description	Default
client_id	Optional[str]	Entra ID app client ID. Falls back to ENTRA_CLIENT_ID.	None
tenant_id	Optional[str]	Entra ID tenant ID. Falls back to ENTRA_TENANT_ID.	None

Returns:

Type	Description
AuthConfig	A fully configured AuthConfig for Entra ID.

Raises:

Type	Description
ValueError	If required configuration is missing.

Source code in src/agora_workbench/code_execution/auth/entra.py

def create_entra_auth_config(
    client_id: Optional[str] = None,
    tenant_id: Optional[str] = None,
) -> AuthConfig:
    """
    Create an AuthConfig using Azure Entra ID for all three auth concerns.

    This is the default configuration for production and local-dev-with-Azure
    deployments. Reads from environment variables if arguments are not provided.

    Args:
        client_id: Entra ID app client ID. Falls back to ``ENTRA_CLIENT_ID``.
        tenant_id: Entra ID tenant ID. Falls back to ``ENTRA_TENANT_ID``.

    Returns:
        A fully configured AuthConfig for Entra ID.

    Raises:
        ValueError: If required configuration is missing.
    """
    resolved_client_id = client_id or os.getenv("ENTRA_CLIENT_ID")
    resolved_tenant_id = tenant_id or os.getenv("ENTRA_TENANT_ID")

    missing = []
    if not resolved_client_id:
        missing.append("ENTRA_CLIENT_ID")
    if not resolved_tenant_id:
        missing.append("ENTRA_TENANT_ID")
    if missing:
        raise ValueError(
            f"Missing required Entra ID configuration: {', '.join(missing)}. "
            "Set via environment variables or pass directly."
        )

    assert resolved_client_id is not None
    assert resolved_tenant_id is not None

    token_validator = EntraTokenValidator(client_id=resolved_client_id, tenant_id=resolved_tenant_id)
    identity_extractor = EntraIdentityExtractor()

    def credential_factory(_user_token: str) -> CredentialProvider:
        return EntraCredentialProvider()

    www_auth = 'Bearer resource_metadata="/.well-known/oauth-protected-resource"'

    # RFC 9728 protected-resource metadata, published through the public
    # AuthConfig contract rather than read off the validator's private state.
    protected_resource_metadata = {
        "resource": f"api://{resolved_client_id}",
        "authorization_servers": [f"https://login.microsoftonline.com/{resolved_tenant_id}/v2.0"],
        "scopes_supported": [f"api://{resolved_client_id}/.default"],
        "bearer_methods_supported": ["header"],
    }

    return AuthConfig(
        token_validator=token_validator,
        identity_extractor=identity_extractor,
        credential_provider_factory=credential_factory,
        www_authenticate_value=www_auth,
        protected_resource_metadata=protected_resource_metadata,
    )