Showcasing allauth IdP: build an MCP server

  INSTALLED_APPS = [
      ...
      # Add these apps
      "allauth",
      "allauth.account",
      "allauth.idp.oidc",
  ]

  MIDDLEWARE = [
      ...
      # Add this middleware
      "allauth.account.middleware.AccountMiddleware",
  ]

  AUTHENTICATION_BACKENDS = [
      "django.contrib.auth.backends.ModelBackend",
      "allauth.account.auth_backends.AuthenticationBackend",
  ]

  urlpatterns = [
      path("admin/", admin.site.urls),
      path("", include("allauth.idp.urls")),
      path("accounts/", include("allauth.urls")),
  ]

  IDP_OIDC_PRIVATE_KEY = """
  -----BEGIN PRIVATE KEY-----
  MIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQCVX7KO32TgyuzK
  ...
  uJR2webdty7MlApkMKe4Y0rQ
  -----END PRIVATE KEY-----
  """

  IDP_OIDC_DCR_ENABLED = True
  IDP_OIDC_DCR_REQUIRES_INITIAL_ACCESS_TOKEN = False

  urlpatterns = [
      ...
      path("", include("myproject.myapp.urls")),
  ]

  from django.urls import path

  from myproject.myapp.views import mcp

  urlpatterns = [
      path('mcp', mcp),
  ]

  """
  Note: For production use, consider one of the available third-party MCP
  packages. We are rolling our own here purely to demystify what happens under the
  hood by providing a bare-bones, white-box implementation.
  """

  from enum import IntEnum
  from functools import wraps
  import json
  from django.urls import reverse

  from django.http import HttpResponse, JsonResponse
  from django.views.decorators.csrf import csrf_exempt
  from allauth.idp.oidc.models import Token


  # Payload returned to the MCP host "as is" when it lists available tools.
  TOOLS = [
      {
          "name": "lost_numbers",
          "description": "Returns the lucky numbers from the TV series Lost.",
          "inputSchema": {"type": "object", "properties": {}},
      }
  ]


  def handle_initialize(params):
      return {
          "protocolVersion": "2024-11-05",
          "capabilities": {"tools": {}},
          "serverInfo": {"name": "myproject", "version": "0.1.0"},
      }


  def handle_tools_list(params):
      return {"tools": TOOLS}


  def handle_tools_call(params):
      if params.get("name") == "lost_numbers":
          # Our one and only demo tool.
          return {
              "content": [
                  {"type": "text", "text": json.dumps([4, 8, 15, 16, 23, 42])}
              ]
          }
      return {"isError": True, "content": [{"type": "text", "text": "Unknown tool"}]}


  METHODS = {
      "initialize": handle_initialize,
      "tools/list": handle_tools_list,
      "tools/call": handle_tools_call,
  }


  class JsonRpcErrorCode(IntEnum):
      PARSE_ERROR = -32700
      INVALID_REQUEST = -32600
      METHOD_NOT_FOUND = -32601


  class JsonRpcErrorResponse(JsonResponse):
      def __init__(self, code: JsonRpcErrorCode, message: str, id=None):
          super().__init__({"jsonrpc": "2.0", "error": {"code": code, "message": message}, "id": id})


  def mcp_authentication_required(view):
      """
      Demo 'authentication-required' decorator. If the token lookup succeeds,
      you may need to perform additional checks for a real world setup (e.g. user,
      scopes, resource).
      """

      @wraps(view)
      def wrapper(request, *args, **kwargs):
          auth = request.headers.get("Authorization", "")
          method, _, raw_token = auth.partition(" ")
          token = None
          if method.lower() == "bearer" and raw_token:
              token = Token.objects.lookup(Token.Type.ACCESS_TOKEN, raw_token)
          if not token:
              resource_url = request.build_absolute_uri(reverse("idp:oidc:configuration"))
              response = HttpResponse(status=401)
              response["WWW-Authenticate"] = f'Bearer resource_metadata="{resource_url}"'
              return response
          # TODO: Insert more ``token`` checks here...
          return view(request, *args, **kwargs)
      return wrapper


  @csrf_exempt
  @mcp_authentication_required
  def mcp(request):
      if request.method != "POST":
          return JsonRpcErrorResponse(JsonRpcErrorCode.INVALID_REQUEST, "POST required")
      try:
          body = json.loads(request.body)
      except json.JSONDecodeError:
          return JsonRpcErrorResponse(JsonRpcErrorCode.PARSE_ERROR, "Parse error")

      method = body.get("method")
      request_id = body.get("id")
      params = body.get("params", {})

      if request_id is None:
          return JsonResponse({}, status=202)

      handler = METHODS.get(method)
      if not handler:
          return JsonRpcErrorResponse(JsonRpcErrorCode.METHOD_NOT_FOUND, "Method not found", id=request_id)

      result = handler(params)
      return JsonResponse({"jsonrpc": "2.0", "result": result, "id": request_id})