Flask

This guide shows a minimal SCIM server built with Flask and scim2_models. It focuses on:

• validate incoming SCIM payloads with the right Context;

• serialize resources and collections as SCIM responses;

• expose responses with the application/scim+json media type;

• convert validation errors into SCIM Error payloads.

The example uses User as a concrete resource type, but the same pattern applies to any other resource such as Group. The storage and mapping layers are defined in the Integrations section and shared across all integration examples. The complete runnable file is available in the Complete example section.

pip install flask scim2-models

Blueprint setup

Start with a Flask blueprint and an after_request() hook that sets the application/scim+json content type on every response.

bp = Blueprint("scim", __name__, url_prefix="/scim/v2")


@bp.after_request
def set_scim_content_type(response):
    """Expose every endpoint with the SCIM media type."""
    response.headers["Content-Type"] = "application/scim+json"
    return response


def resource_location(app_record):
    """Return the canonical URL for a user record."""
    return url_for("scim.get_user", app_record=app_record, _external=True)

Optional Flask refinements

The core SCIM flow only needs the blueprint and the endpoints below. Flask also offers a few convenient integration patterns that can keep the views shorter and help keep framework-level errors aligned with SCIM responses.

Flask converters

Converters let Flask resolve route parameters before the view function is called. Define one converter per resource type: it maps a resource identifier to an application record and lets Flask handle the not-found case before entering the view. bp.record_once registers the converter on the app when the blueprint is attached.

class UserConverter(BaseConverter):
    """Resolve a user identifier to an application record."""

    def to_python(self, id):
        try:
            return get_record(id)
        except KeyError:
            raise RoutingValidationError()

    def to_url(self, record):
        return record["id"]


@bp.record_once
def _register_converter(state):
    state.app.url_map.converters["user"] = UserConverter

Error handlers

The error handlers keep Pydantic validation errors and HTTP errors aligned with SCIM responses.

@bp.errorhandler(ValidationError)
def handle_validation_error(error):
    """Turn Pydantic validation errors into SCIM error responses."""
    scim_error = Error.from_validation_error(error.errors()[0])
    return scim_error.model_dump(), scim_error.status


@bp.errorhandler(HTTPException)
def handle_http_error(error):
    """Turn HTTP errors into SCIM error responses."""
    scim_error = Error(status=error.code, detail=str(error.description))
    return scim_error.model_dump(), error.code


@bp.errorhandler(SCIMException)
def handle_scim_error(error):
    """Turn SCIM exceptions into SCIM error responses."""
    scim_error = error.to_error()
    return scim_error.model_dump(), scim_error.status

If model_validate() fails, Flask routes the ValidationError to handle_validation_error and the client receives a SCIM Error response. handle_scim_error catches any SCIMException (uniqueness, mutability, …) and returns the appropriate SCIM Error response. handle_http_error catches any HTTPException (404, 412, 405, …) and returns the corresponding SCIM Error response.

Endpoints

The routes below serve /Users, but the same structure applies to any resource type: replace the mapping helpers, the model class, and the URL prefix to expose /Groups or any other collection.

GET /Users/<id>

Parse query parameters with ResponseParameters, convert the native record to a SCIM resource with your mapping helper, then serialize with RESOURCE_QUERY_RESPONSE, forwarding req.attributes and req.excluded_attributes so the response only includes the requested fields.

@bp.get("/Users/<user:app_record>")
def get_user(app_record):
    """Return one SCIM user."""
    req = ResponseParameters.model_validate(request.args.to_dict())
    scim_user = to_scim_user(app_record, resource_location(app_record))
    return scim_user.model_dump(
        scim_ctx=Context.RESOURCE_QUERY_RESPONSE,
        attributes=req.attributes,
        excluded_attributes=req.excluded_attributes,
    )

DELETE /Users/<id>

Remove the record from the store and return an empty 204 response. No SCIM serialization is needed.

@bp.delete("/Users/<user:app_record>")
def delete_user(app_record):
    """Delete an existing user."""
    delete_record(app_record["id"])
    return "", HTTPStatus.NO_CONTENT

PATCH /Users/<id>

Validate the patch payload with RESOURCE_PATCH_REQUEST, apply it to a SCIM conversion of the native record with patch(), convert back to native and persist, then serialize the result with RESOURCE_PATCH_RESPONSE. PatchOp is generic and works with any resource type.

@bp.patch("/Users/<user:app_record>")
def patch_user(app_record):
    """Apply a SCIM PatchOp to an existing user."""
    req = ResponseParameters.model_validate(request.args.to_dict())
    scim_user = to_scim_user(app_record, resource_location(app_record))
    patch = PatchOp[User].model_validate(
        request.get_json(),
        scim_ctx=Context.RESOURCE_PATCH_REQUEST,
    )
    patch.patch(scim_user)

    updated_record = from_scim_user(scim_user)
    save_record(updated_record)

    return scim_user.model_dump(
        scim_ctx=Context.RESOURCE_PATCH_RESPONSE,
        attributes=req.attributes,
        excluded_attributes=req.excluded_attributes,
    )

PUT /Users/<id>

Validate the full replacement payload with RESOURCE_REPLACEMENT_REQUEST, then call replace() to verify that immutable attributes have not been modified.

@bp.put("/Users/<user:app_record>")
def replace_user(app_record):
    """Replace an existing user with a full SCIM resource."""
    req = ResponseParameters.model_validate(request.args.to_dict())
    existing_user = to_scim_user(app_record, resource_location(app_record))
    replacement = User.model_validate(
        request.get_json(),
        scim_ctx=Context.RESOURCE_REPLACEMENT_REQUEST,
    )
    replacement.replace(existing_user)

    updated_record = from_scim_user(replacement)
    save_record(updated_record)

    response_user = to_scim_user(updated_record, resource_location(updated_record))
    return response_user.model_dump(
        scim_ctx=Context.RESOURCE_REPLACEMENT_RESPONSE,
        attributes=req.attributes,
        excluded_attributes=req.excluded_attributes,
    )

GET /Users

Parse pagination and filtering parameters with SearchRequest, slice the store accordingly, then wrap the page in a ListResponse serialized with RESOURCE_QUERY_RESPONSE. Pass req.attributes and req.excluded_attributes to model_dump_json() so that the attributes and excludedAttributes query parameters are applied to each embedded resource.

@bp.get("/Users")
def list_users():
    """Return one page of users as a SCIM ListResponse."""
    req = SearchRequest.model_validate(request.args.to_dict())
    total, page = list_records(req.start_index_0, req.stop_index_0)
    resources = [to_scim_user(record, resource_location(record)) for record in page]
    response = ListResponse[User](
        total_results=total,
        start_index=req.start_index or 1,
        items_per_page=len(resources),
        resources=resources,
    )
    return response.model_dump(
        scim_ctx=Context.RESOURCE_QUERY_RESPONSE,
        attributes=req.attributes,
        excluded_attributes=req.excluded_attributes,
    )

POST /Users

Validate the creation payload with RESOURCE_CREATION_REQUEST, convert to native and persist, then serialize the created resource with RESOURCE_CREATION_RESPONSE.

@bp.post("/Users")
def create_user():
    """Validate a SCIM creation payload and store the new user."""
    req = ResponseParameters.model_validate(request.args.to_dict())
    request_user = User.model_validate(
        request.get_json(),
        scim_ctx=Context.RESOURCE_CREATION_REQUEST,
    )
    app_record = from_scim_user(request_user)
    save_record(app_record)

    response_user = to_scim_user(app_record, resource_location(app_record))
    return (
        response_user.model_dump(
            scim_ctx=Context.RESOURCE_CREATION_RESPONSE,
            attributes=req.attributes,
            excluded_attributes=req.excluded_attributes,
        ),
        HTTPStatus.CREATED,
    )

Resource versioning (ETags)

SCIM supports resource versioning through HTTP ETags (RFC 7644 §3.14). Both ETag checks are handled centrally rather than in individual endpoints:

• The after_request hook extracts meta.version from the response body and sets the ETag response header. Werkzeug’s make_conditional() handles If-None-Match to return a 304 Not Modified when the client already has the current version.

• The before_request hook reads the If-Match header on write operations (PUT, PATCH, DELETE) and raises PreconditionFailed on mismatch, since make_conditional() only acts on GET/HEAD.

make_etag computes a weak ETag from each record and populates version.

@bp.after_request
def set_etag_header(response):
    """Extract ``ETag`` from ``meta.version`` and handle conditional responses."""
    data = response.get_json(silent=True)
    if meta := (data or {}).get("meta"):
        if version := meta.get("version"):
            response.headers["ETag"] = version
    response.make_conditional(request)
    return response


@bp.before_request
def check_etag():
    """Verify ``If-Match`` on write operations.

    :raises ~werkzeug.exceptions.PreconditionFailed: If the header is present and does not match.
    """
    if request.method not in ("PUT", "PATCH", "DELETE"):
        return
    app_record = request.view_args.get("app_record")
    if app_record is None:
        return
    if_match = request.headers.get("If-Match")
    if not if_match:
        return
    if if_match.strip() == "*":
        return
    etag = make_etag(app_record)
    tags = [t.strip() for t in if_match.split(",")]
    if etag not in tags:
        raise PreconditionFailed("ETag mismatch")

Tip

If your application uses SQLAlchemy, the built-in version counter can serve as ETag value directly, removing the need for a manual hash.

Discovery endpoints

SCIM defines three read-only endpoints that let clients discover the server’s capabilities and the resources it exposes (RFC 7644 §4). The shared discovery helpers that build Schema, ResourceType and ServiceProviderConfig objects are defined in the Integrations section.

GET /Schemas and GET /Schemas/<id>

Return all Schema objects or look one up by its URI. Schemas are built automatically from resource models with to_schema(). The collection endpoint parses pagination parameters with SearchRequest, following the same pattern as GET /Users.

@bp.get("/Schemas")
def list_schemas():
    """Return one page of SCIM schemas the server exposes."""
    req = SearchRequest.model_validate(request.args.to_dict())
    total, page = get_schemas(req.start_index_0, req.stop_index_0)
    response = ListResponse[Schema](
        total_results=total,
        start_index=req.start_index or 1,
        items_per_page=len(page),
        resources=page,
    )
    return response.model_dump(scim_ctx=Context.RESOURCE_QUERY_RESPONSE)


@bp.get("/Schemas/<path:schema_id>")
def get_schema_by_id(schema_id):
    """Return one SCIM schema by its URI identifier."""
    try:
        schema = get_schema(schema_id)
    except KeyError:
        scim_error = Error(status=404, detail=f"Schema {schema_id!r} not found")
        return scim_error.model_dump(), HTTPStatus.NOT_FOUND
    return schema.model_dump(scim_ctx=Context.RESOURCE_QUERY_RESPONSE)

GET /ResourceTypes and GET /ResourceTypes/<id>

Return all ResourceType objects or look one up by its identifier. Resource types are built automatically from resource models with from_resource(). The collection endpoint parses pagination parameters with SearchRequest, following the same pattern as GET /Users.

@bp.get("/ResourceTypes")
def list_resource_types():
    """Return one page of SCIM resource types the server exposes."""
    req = SearchRequest.model_validate(request.args.to_dict())
    total, page = get_resource_types(req.start_index_0, req.stop_index_0)
    response = ListResponse[ResourceType](
        total_results=total,
        start_index=req.start_index or 1,
        items_per_page=len(page),
        resources=page,
    )
    return response.model_dump(scim_ctx=Context.RESOURCE_QUERY_RESPONSE)


@bp.get("/ResourceTypes/<resource_type_id>")
def get_resource_type_by_id(resource_type_id):
    """Return one SCIM resource type by its identifier."""
    try:
        rt = get_resource_type(resource_type_id)
    except KeyError:
        scim_error = Error(
            status=404, detail=f"ResourceType {resource_type_id!r} not found"
        )
        return scim_error.model_dump(), HTTPStatus.NOT_FOUND
    return rt.model_dump(scim_ctx=Context.RESOURCE_QUERY_RESPONSE)

GET /ServiceProviderConfig

Return the ServiceProviderConfig singleton that describes the features the server supports (patch, bulk, filtering, etc.).

@bp.get("/ServiceProviderConfig")
def get_service_provider_config():
    """Return the SCIM service provider configuration."""
    return service_provider_config.model_dump(
        scim_ctx=Context.RESOURCE_QUERY_RESPONSE
    )

Complete example

from http import HTTPStatus

from flask import Blueprint
from flask import request
from flask import url_for
from pydantic import ValidationError
from werkzeug.exceptions import HTTPException
from werkzeug.exceptions import PreconditionFailed
from werkzeug.routing import BaseConverter
from werkzeug.routing import ValidationError as RoutingValidationError

from scim2_models import Context
from scim2_models import Error
from scim2_models import ListResponse
from scim2_models import PatchOp
from scim2_models import ResourceType
from scim2_models import ResponseParameters
from scim2_models import Schema
from scim2_models import SCIMException
from scim2_models import SearchRequest
from scim2_models import User

from .integrations import delete_record
from .integrations import from_scim_user
from .integrations import get_record
from .integrations import get_resource_type
from .integrations import get_resource_types
from .integrations import get_schema
from .integrations import get_schemas
from .integrations import list_records
from .integrations import make_etag
from .integrations import save_record
from .integrations import service_provider_config
from .integrations import to_scim_user

# -- setup-start --
bp = Blueprint("scim", __name__, url_prefix="/scim/v2")


@bp.after_request
def set_scim_content_type(response):
    """Expose every endpoint with the SCIM media type."""
    response.headers["Content-Type"] = "application/scim+json"
    return response


def resource_location(app_record):
    """Return the canonical URL for a user record."""
    return url_for("scim.get_user", app_record=app_record, _external=True)
# -- setup-end --


# -- etag-start --
@bp.after_request
def set_etag_header(response):
    """Extract ``ETag`` from ``meta.version`` and handle conditional responses."""
    data = response.get_json(silent=True)
    if meta := (data or {}).get("meta"):
        if version := meta.get("version"):
            response.headers["ETag"] = version
    response.make_conditional(request)
    return response


@bp.before_request
def check_etag():
    """Verify ``If-Match`` on write operations.

    :raises ~werkzeug.exceptions.PreconditionFailed: If the header is present and does not match.
    """
    if request.method not in ("PUT", "PATCH", "DELETE"):
        return
    app_record = request.view_args.get("app_record")
    if app_record is None:
        return
    if_match = request.headers.get("If-Match")
    if not if_match:
        return
    if if_match.strip() == "*":
        return
    etag = make_etag(app_record)
    tags = [t.strip() for t in if_match.split(",")]
    if etag not in tags:
        raise PreconditionFailed("ETag mismatch")
# -- etag-end --


# -- refinements-start --
# -- converters-start --
class UserConverter(BaseConverter):
    """Resolve a user identifier to an application record."""

    def to_python(self, id):
        try:
            return get_record(id)
        except KeyError:
            raise RoutingValidationError()

    def to_url(self, record):
        return record["id"]


@bp.record_once
def _register_converter(state):
    state.app.url_map.converters["user"] = UserConverter
# -- converters-end --


# -- error-handlers-start --
@bp.errorhandler(ValidationError)
def handle_validation_error(error):
    """Turn Pydantic validation errors into SCIM error responses."""
    scim_error = Error.from_validation_error(error.errors()[0])
    return scim_error.model_dump(), scim_error.status


@bp.errorhandler(HTTPException)
def handle_http_error(error):
    """Turn HTTP errors into SCIM error responses."""
    scim_error = Error(status=error.code, detail=str(error.description))
    return scim_error.model_dump(), error.code


@bp.errorhandler(SCIMException)
def handle_scim_error(error):
    """Turn SCIM exceptions into SCIM error responses."""
    scim_error = error.to_error()
    return scim_error.model_dump(), scim_error.status
# -- error-handlers-end --
# -- refinements-end --


# -- endpoints-start --
# -- single-resource-start --
# -- get-user-start --
@bp.get("/Users/<user:app_record>")
def get_user(app_record):
    """Return one SCIM user."""
    req = ResponseParameters.model_validate(request.args.to_dict())
    scim_user = to_scim_user(app_record, resource_location(app_record))
    return scim_user.model_dump(
        scim_ctx=Context.RESOURCE_QUERY_RESPONSE,
        attributes=req.attributes,
        excluded_attributes=req.excluded_attributes,
    )
# -- get-user-end --


# -- patch-user-start --
@bp.patch("/Users/<user:app_record>")
def patch_user(app_record):
    """Apply a SCIM PatchOp to an existing user."""
    req = ResponseParameters.model_validate(request.args.to_dict())
    scim_user = to_scim_user(app_record, resource_location(app_record))
    patch = PatchOp[User].model_validate(
        request.get_json(),
        scim_ctx=Context.RESOURCE_PATCH_REQUEST,
    )
    patch.patch(scim_user)

    updated_record = from_scim_user(scim_user)
    save_record(updated_record)

    return scim_user.model_dump(
        scim_ctx=Context.RESOURCE_PATCH_RESPONSE,
        attributes=req.attributes,
        excluded_attributes=req.excluded_attributes,
    )
# -- patch-user-end --


# -- put-user-start --
@bp.put("/Users/<user:app_record>")
def replace_user(app_record):
    """Replace an existing user with a full SCIM resource."""
    req = ResponseParameters.model_validate(request.args.to_dict())
    existing_user = to_scim_user(app_record, resource_location(app_record))
    replacement = User.model_validate(
        request.get_json(),
        scim_ctx=Context.RESOURCE_REPLACEMENT_REQUEST,
    )
    replacement.replace(existing_user)

    updated_record = from_scim_user(replacement)
    save_record(updated_record)

    response_user = to_scim_user(updated_record, resource_location(updated_record))
    return response_user.model_dump(
        scim_ctx=Context.RESOURCE_REPLACEMENT_RESPONSE,
        attributes=req.attributes,
        excluded_attributes=req.excluded_attributes,
    )
# -- put-user-end --


# -- delete-user-start --
@bp.delete("/Users/<user:app_record>")
def delete_user(app_record):
    """Delete an existing user."""
    delete_record(app_record["id"])
    return "", HTTPStatus.NO_CONTENT
# -- delete-user-end --
# -- single-resource-end --


# -- collection-start --
# -- list-users-start --
@bp.get("/Users")
def list_users():
    """Return one page of users as a SCIM ListResponse."""
    req = SearchRequest.model_validate(request.args.to_dict())
    total, page = list_records(req.start_index_0, req.stop_index_0)
    resources = [to_scim_user(record, resource_location(record)) for record in page]
    response = ListResponse[User](
        total_results=total,
        start_index=req.start_index or 1,
        items_per_page=len(resources),
        resources=resources,
    )
    return response.model_dump(
        scim_ctx=Context.RESOURCE_QUERY_RESPONSE,
        attributes=req.attributes,
        excluded_attributes=req.excluded_attributes,
    )
# -- list-users-end --


# -- create-user-start --
@bp.post("/Users")
def create_user():
    """Validate a SCIM creation payload and store the new user."""
    req = ResponseParameters.model_validate(request.args.to_dict())
    request_user = User.model_validate(
        request.get_json(),
        scim_ctx=Context.RESOURCE_CREATION_REQUEST,
    )
    app_record = from_scim_user(request_user)
    save_record(app_record)

    response_user = to_scim_user(app_record, resource_location(app_record))
    return (
        response_user.model_dump(
            scim_ctx=Context.RESOURCE_CREATION_RESPONSE,
            attributes=req.attributes,
            excluded_attributes=req.excluded_attributes,
        ),
        HTTPStatus.CREATED,
    )
# -- create-user-end --
# -- collection-end --


# -- discovery-start --
# -- schemas-start --
@bp.get("/Schemas")
def list_schemas():
    """Return one page of SCIM schemas the server exposes."""
    req = SearchRequest.model_validate(request.args.to_dict())
    total, page = get_schemas(req.start_index_0, req.stop_index_0)
    response = ListResponse[Schema](
        total_results=total,
        start_index=req.start_index or 1,
        items_per_page=len(page),
        resources=page,
    )
    return response.model_dump(scim_ctx=Context.RESOURCE_QUERY_RESPONSE)


@bp.get("/Schemas/<path:schema_id>")
def get_schema_by_id(schema_id):
    """Return one SCIM schema by its URI identifier."""
    try:
        schema = get_schema(schema_id)
    except KeyError:
        scim_error = Error(status=404, detail=f"Schema {schema_id!r} not found")
        return scim_error.model_dump(), HTTPStatus.NOT_FOUND
    return schema.model_dump(scim_ctx=Context.RESOURCE_QUERY_RESPONSE)
# -- schemas-end --


# -- resource-types-start --
@bp.get("/ResourceTypes")
def list_resource_types():
    """Return one page of SCIM resource types the server exposes."""
    req = SearchRequest.model_validate(request.args.to_dict())
    total, page = get_resource_types(req.start_index_0, req.stop_index_0)
    response = ListResponse[ResourceType](
        total_results=total,
        start_index=req.start_index or 1,
        items_per_page=len(page),
        resources=page,
    )
    return response.model_dump(scim_ctx=Context.RESOURCE_QUERY_RESPONSE)


@bp.get("/ResourceTypes/<resource_type_id>")
def get_resource_type_by_id(resource_type_id):
    """Return one SCIM resource type by its identifier."""
    try:
        rt = get_resource_type(resource_type_id)
    except KeyError:
        scim_error = Error(
            status=404, detail=f"ResourceType {resource_type_id!r} not found"
        )
        return scim_error.model_dump(), HTTPStatus.NOT_FOUND
    return rt.model_dump(scim_ctx=Context.RESOURCE_QUERY_RESPONSE)
# -- resource-types-end --


# -- service-provider-config-start --
@bp.get("/ServiceProviderConfig")
def get_service_provider_config():
    """Return the SCIM service provider configuration."""
    return service_provider_config.model_dump(
        scim_ctx=Context.RESOURCE_QUERY_RESPONSE
    )
# -- service-provider-config-end --
# -- discovery-end --
# -- endpoints-end --