Advertise the authorization server's full scope catalog + 401 scope hint

.gitignore

@@ -12,4 +12,7 @@ wheels/
 # VSCode
 .vscode
 
-.env
+.env
+# Playwright MCP artifacts
+.playwright-mcp/
+.DS_Store

README.md

@@ -11,6 +11,8 @@ through your browser. The first time you connect, your client opens an Appwrite
 consent screen; approve the scopes and you're connected. There are no keys to
 copy.
 
+![How the Appwrite MCP server handles OAuth and scopes](docs/appwrite-mcp-flow.svg)
+
 ## Connect your client
 
 Pick your client below. Each adds the hosted Appwrite Cloud server.

src/mcp_server_appwrite/auth.py

@@ -158,17 +158,21 @@ def authorization_server_metadata_sync() -> dict:
 
 
 def _advertised_scopes(metadata: dict) -> list[str]:
-    """The scope set to advertise: the preferred scopes intersected with the
-    authorization server's live ``scopes_supported`` (so a renamed/removed scope
-    is never advertised). Falls back to mirroring the full discovery list when
-    none of the preferred scopes exist — e.g. a self-hosted project with a
-    custom, compact scope catalog."""
+    """The scope set to advertise. By default (no curated list) this mirrors the
+    authorization server's full live ``scopes_supported`` catalog — clients then
+    request everything and consent-time narrowing is the control point. When a
+    curated list is configured (``MCP_OAUTH_SCOPES``), it is intersected with
+    the discovered catalog so a renamed/removed scope is never advertised,
+    falling back to the full discovered list when the intersection is empty."""
     discovered = metadata.get("scopes_supported")
     if not isinstance(discovered, list):
         raise ValueError(
             f"authorization server discovery missing scopes_supported: {discovery_url()}"
         )
-    scopes = [scope for scope in preferred_scopes() if scope in discovered]
+    preferred = preferred_scopes()
+    if not preferred:
+        return discovered
+    scopes = [scope for scope in preferred if scope in discovered]
     if scopes:
         return scopes
     _log(

src/mcp_server_appwrite/constants.py

@@ -50,12 +50,11 @@
 
 DEFAULT_PROJECT_ID = "console"
 
-PREFERRED_SCOPES = [
-    "openid",
-    "profile",
-    "email",
-    "all",
-]
+# Curated scope allowlist. Empty means "no curation": the MCP mirrors the
+# authorization server's full ``scopes_supported`` catalog so clients request
+# every granular scope and the consent screen becomes the narrowing control
+# point. A deployment can still pin a curated set via ``MCP_OAUTH_SCOPES``.
+PREFERRED_SCOPES: list[str] = []
 
 DISCOVERY_TTL_SECONDS = 300.0
 

src/mcp_server_appwrite/context.py

@@ -92,6 +92,30 @@ def get_appwrite_context(
                 )
             )
 
+    if not projects:
+        # Per-organization project listings need organization-tier scopes; a
+        # grant narrowed to project scopes only still enumerates its bound
+        # projects through the accessible-resources endpoint.
+        for accessible in _accessible_resource_ids(console_client, "projects"):
+            accessible_id = accessible.get("$id")
+            if not isinstance(accessible_id, str) or not accessible_id:
+                continue
+            if project_id and accessible_id != project_id:
+                continue
+            project_client = client_factory(accessible_id, organization_id)
+            project = _get_current_project(project_client, accessible_id) or {
+                "$id": accessible_id
+            }
+            projects.append(
+                _project_context(
+                    project,
+                    project_client,
+                    organization_id=organization_id,
+                    include_services=include_services,
+                    sample_limit=sample_limit,
+                )
+            )
+
     if project_id and not projects:
         project_client = client_factory(project_id, organization_id)
         project = _get_current_project(project_client, project_id) or {
@@ -153,11 +177,45 @@ def _get_current_project(client: Client, project_id: str) -> dict[str, Any] | No
     return {"$id": project_id, "error": result.error} if not result.ok else None
 
 
+def _accessible_resource_ids(client: Client, resource: str) -> list[dict[str, Any]]:
+    """Fallback discovery via the OAuth2 accessible-resources endpoints
+    (``/oauth2/<project>/organizations`` / ``/oauth2/<project>/projects``).
+    Their scopes are granted on every token, so a consent-narrowed grant (no
+    console-wide ``all``) can still enumerate exactly the resources it is
+    bound to. Returns id-only documents."""
+    project_id = client.get_config("project") or ""
+    if not project_id:
+        return []
+    result = _safe_call(client, "get", f"/oauth2/{project_id}/{resource}")
+    if not result.ok or not isinstance(result.value, dict):
+        return []
+    items = result.value.get(resource)
+    if not isinstance(items, list):
+        return []
+    return [
+        {"$id": item["$id"]}
+        for item in items
+        if isinstance(item, dict) and isinstance(item.get("$id"), str)
+    ]
+
+
 def _list_organizations(
     client: Client, organization_id: str | None
 ) -> list[dict[str, Any]]:
     result = _safe_call(client, "get", "/organizations")
     if not result.ok or not isinstance(result.value, dict):
+        # The console-wide organizations listing needs scopes only the `all`
+        # grant carries; a narrowed grant falls back to the accessible-
+        # resources enumeration (ids only, names resolved by later calls).
+        accessible = _accessible_resource_ids(client, "organizations")
+        if accessible:
+            if organization_id:
+                return [
+                    organization
+                    for organization in accessible
+                    if organization.get("$id") == organization_id
+                ]
+            return accessible
         return (
             [{"$id": organization_id, "error": result.error}] if organization_id else []
         )

src/mcp_server_appwrite/http_app.py

@@ -62,6 +62,15 @@ def _log(message: str) -> None:
     print(f"[appwrite-mcp][http] {message}", file=sys.stderr, flush=True)
 
 
+def _is_valid_scope_token(scope: str) -> bool:
+    """RFC 6749 §3.3 scope-token grammar: 1*( %x21 / %x23-5B / %x5D-7E ) —
+    printable ASCII minus space, double quote, and backslash."""
+    return bool(scope) and all(
+        char == "\x21" or "\x23" <= char <= "\x5b" or "\x5d" <= char <= "\x7e"
+        for char in scope
+    )
+
+
 async def _send_401(send: Send) -> None:
     """RFC 9728 §5.1 — 401 with a WWW-Authenticate pointing to resource metadata."""
     metadata_url = resource_metadata_url()
@@ -70,6 +79,24 @@ async def _send_401(send: Send) -> None:
         'error_description="Authentication required"',
         f'resource_metadata="{metadata_url}"',
     ]
+    # MCP spec 2025-11-25 (SEP-835): clients treat the `scope` parameter as
+    # authoritative for what to request. Sourced from the same discovery-backed
+    # metadata as /.well-known; omitted entirely if discovery is unavailable —
+    # an unauthenticated 401 must never turn into a 500. The values land inside
+    # a quoted-string header, so anything outside the RFC 6749 §3.3 scope-token
+    # grammar (quotes, backslashes, control characters such as CRLF) is dropped
+    # rather than injected into the response head.
+    try:
+        scopes = (await protected_resource_metadata()).get("scopes_supported") or []
+        scopes = [
+            scope
+            for scope in scopes
+            if isinstance(scope, str) and _is_valid_scope_token(scope)
+        ]
+        if scopes:
+            parts.append(f'scope="{" ".join(scopes)}"')
+    except Exception:
+        pass
     www_authenticate = f"Bearer {', '.join(parts)}"
 
     body = json.dumps(

tests/unit/test_auth.py

@@ -61,47 +61,32 @@ def test_advertised_scopes_use_cache_without_network(self):
             scopes = asyncio.run(auth.protected_resource_metadata())["scopes_supported"]
         finally:
             auth._discovery_cache.pop(pid, None)
-        # None of the preferred scopes exist, so the full discovered list is
-        # mirrored (custom scope catalogs on self-hosted projects).
+        # No curated list configured, so the discovered catalog is mirrored.
         self.assertEqual(scopes, ["rows.read", "rows.write"])
 
-    def test_advertised_scopes_prefer_curated_subset(self):
-        # The Cloud console authorization server advertises a very large
-        # fine-grained scope catalog; the MCP must advertise only the compact
-        # preferred set (clients request every advertised scope, and the
-        # authorize endpoint caps the scope parameter length).
+    def test_advertised_scopes_mirror_full_catalog_by_default(self):
+        # The MCP mirrors the authorization server's full scope catalog so
+        # clients request everything and consent-time narrowing is the control
+        # point (granular MCP scopes design).
+        catalog = [
+            "openid",
+            "profile",
+            "email",
+            "phone",
+            "all",
+            "project:all",
+            "organization:all",
+            "project:users.read",
+            "project:users.write",
+            "organization:projects.read",
+        ]
         pid = auth.configured_project_id()
-        auth._store_discovery(
-            pid,
-            discovery_doc(
-                [
-                    "openid",
-                    "profile",
-                    "email",
-                    "phone",
-                    "all",
-                    "project:all",
-                    "organization:all",
-                    "project:users.read",
-                    "project:users.write",
-                    "organization:projects.read",
-                ]
-            ),
-        )
-        try:
-            scopes = asyncio.run(auth.protected_resource_metadata())["scopes_supported"]
-        finally:
-            auth._discovery_cache.pop(pid, None)
-        self.assertEqual(scopes, ["openid", "profile", "email", "all"])
-
-    def test_advertised_scopes_drop_preferred_scopes_missing_from_discovery(self):
-        pid = auth.configured_project_id()
-        auth._store_discovery(pid, discovery_doc(["openid", "email", "all"]))
+        auth._store_discovery(pid, discovery_doc(catalog))
         try:
             scopes = asyncio.run(auth.protected_resource_metadata())["scopes_supported"]
         finally:
             auth._discovery_cache.pop(pid, None)
-        self.assertEqual(scopes, ["openid", "email", "all"])
+        self.assertEqual(scopes, catalog)
 
     def test_advertised_scopes_env_override(self):
         pid = auth.configured_project_id()
@@ -119,6 +104,46 @@ def test_advertised_scopes_env_override(self):
             auth._discovery_cache.pop(pid, None)
         self.assertEqual(scopes, ["openid", "project:all"])
 
+    def test_advertised_scopes_env_override_drops_undiscovered_scopes(self):
+        # A curated scope missing from the live catalog is never advertised.
+        pid = auth.configured_project_id()
+        auth._store_discovery(pid, discovery_doc(["openid", "email", "all"]))
+        try:
+            with mock.patch.dict(
+                os.environ, {"MCP_OAUTH_SCOPES": "openid email all phone"}
+            ):
+                scopes = asyncio.run(auth.protected_resource_metadata())[
+                    "scopes_supported"
+                ]
+        finally:
+            auth._discovery_cache.pop(pid, None)
+        self.assertEqual(scopes, ["openid", "email", "all"])
+
+    def test_advertised_scopes_env_override_falls_back_to_full_catalog(self):
+        # When none of the curated scopes exist in the live catalog, the full
+        # discovered list is mirrored instead of advertising nothing.
+        pid = auth.configured_project_id()
+        auth._store_discovery(pid, discovery_doc(["rows.read", "rows.write"]))
+        try:
+            with mock.patch.dict(os.environ, {"MCP_OAUTH_SCOPES": "openid all"}):
+                scopes = asyncio.run(auth.protected_resource_metadata())[
+                    "scopes_supported"
+                ]
+        finally:
+            auth._discovery_cache.pop(pid, None)
+        self.assertEqual(scopes, ["rows.read", "rows.write"])
+
+    def test_advertised_scopes_raise_when_discovery_missing_scopes_supported(self):
+        pid = auth.configured_project_id()
+        doc = discovery_doc([])
+        del doc["scopes_supported"]
+        auth._store_discovery(pid, doc)
+        try:
+            with self.assertRaises(ValueError):
+                asyncio.run(auth.protected_resource_metadata())
+        finally:
+            auth._discovery_cache.pop(pid, None)
+
     def test_discovery_cache_expires_after_ttl(self):
         pid = auth.configured_project_id()
         auth._store_discovery(pid, {"issuer": "x", "jwks_uri": "y"})

tests/unit/test_context.py

@@ -1,6 +1,7 @@
 import unittest
 
 from appwrite.client import Client
+from appwrite.exception import AppwriteException
 
 from mcp_server_appwrite.context import get_appwrite_context
 from mcp_server_appwrite.server import _get_context_for_request
@@ -116,6 +117,63 @@ def factory(project_id, organization_id):
         self.assertIn((None, "org-1"), seen)
         self.assertIn(("project-1", "org-1"), seen)
 
+    def test_oauth_context_falls_back_to_accessible_resources_when_narrowed(self):
+        # A consent-narrowed grant (no console-wide `all`) cannot use the
+        # console-tier /organizations and per-org project listings; discovery
+        # must fall back to the always-granted OAuth2 accessible-resources
+        # endpoints and still surface the bound organization and project.
+        def denied(_params):
+            raise AppwriteException("missing scopes", 401, "general_unauthorized_scope")
+
+        console = FakeClient(
+            {
+                ("get", "/account"): denied,
+                ("get", "/organizations"): denied,
+                ("get", "/oauth2/console/organizations"): {
+                    "total": 1,
+                    "organizations": [{"$id": "org-1"}],
+                },
+                ("get", "/oauth2/console/projects"): {
+                    "total": 1,
+                    "projects": [{"$id": "project-a"}],
+                },
+            },
+            project="console",
+        )
+        org = FakeClient({("get", "/organization/projects"): denied})
+        project = FakeClient(
+            {
+                ("get", "/project"): {"$id": "project-a", "name": "Project A"},
+                ("get", "/tablesdb"): {
+                    "total": 1,
+                    "databases": [{"$id": "db-1", "name": "Main"}],
+                },
+                ("get", "/storage/buckets"): denied,
+            },
+            project="project-a",
+        )
+
+        def factory(project_id, organization_id):
+            if project_id:
+                return project
+            if organization_id:
+                return org
+            return console
+
+        context = get_appwrite_context(
+            console,
+            mode="oauth_console",
+            client_factory=factory,
+        )
+
+        self.assertIn("error", context["account"])
+        self.assertEqual(context["organizations"], [{"$id": "org-1"}])
+        self.assertEqual(context["projects"][0]["$id"], "project-a")
+        self.assertEqual(context["projects"][0]["name"], "Project A")
+        self.assertEqual(context["projects"][0]["services"]["tablesdb"]["total"], 1)
+        # Ungranted probes surface their scope error instead of failing the call.
+        self.assertIn("error", context["projects"][0]["services"]["storage"])
+
     def test_compact_output_preserves_false_resource_state(self):
         client = FakeClient(
             {