Route project-scoped calls to the project's home region

src/mcp_server_appwrite/auth.py

@@ -27,9 +27,9 @@
 
 from . import telemetry
 from .constants import (
+    CACHE_TTL_SECONDS,
     DEFAULT_ENDPOINT,
     DEFAULT_PROJECT_ID,
-    DISCOVERY_TTL_SECONDS,
     PREFERRED_SCOPES,
 )
 
@@ -89,7 +89,7 @@ def _cached_discovery(project_id: str, *, allow_stale: bool = False) -> dict | N
     if entry is None:
         return None
     fetched_at, document = entry
-    if allow_stale or time.monotonic() - fetched_at < DISCOVERY_TTL_SECONDS:
+    if allow_stale or time.monotonic() - fetched_at < CACHE_TTL_SECONDS:
         return document
     return None
 

src/mcp_server_appwrite/constants.py

@@ -17,6 +17,8 @@
 SERVER_VERSION = "0.8.3"
 
 DEFAULT_ENDPOINT = "https://cloud.appwrite.io/v1"
+# Region reported by single-region deployments; carries no region subdomain.
+DEFAULT_REGION = "default"
 DEFAULT_TRANSPORT = "stdio"
 TRANSPORTS = {"stdio", "http"}
 VALIDATION_SERVICE_ORDER = (
@@ -57,7 +59,8 @@
     "all",
 ]
 
-DISCOVERY_TTL_SECONDS = 300.0
+# Shared TTL for cached upstream lookups (OAuth discovery, project regions).
+CACHE_TTL_SECONDS = 300.0
 
 # --- http_app -------------------------------------------------------------
 

src/mcp_server_appwrite/server.py

@@ -22,7 +22,7 @@
 from pathlib import Path
 from types import UnionType
 from typing import Any, Union, get_args, get_origin
-from urllib.parse import unquote, urlsplit
+from urllib.parse import unquote, urlsplit, urlunsplit
 
 import httpx
 import mcp.server.stdio
@@ -41,8 +41,10 @@
 
 from . import telemetry
 from .constants import (
+    CACHE_TTL_SECONDS,
     CATALOG_URI,
     DEFAULT_ENDPOINT,
+    DEFAULT_REGION,
     DEFAULT_TRANSPORT,
     EXCLUDED_SERVICES,
     FETCH_MAX_REDIRECTS,
@@ -221,14 +223,79 @@ def build_client_for_request(
     return client
 
 
+# Appwrite Cloud is multi-region: project metadata is global (any gateway can
+# list projects), but a project's data plane lives only in its home region and
+# must be addressed through the region subdomain (e.g.
+# ``https://sgp.cloud.appwrite.io/v1``); other gateways reject project-scoped
+# calls with 401 general_access_forbidden. Regions are looked up once per
+# project via the console API and cached briefly; the 'default' sentinel is
+# cached like any region, so a project that migrates regions may be routed to
+# its old home for up to CACHE_TTL_SECONDS.
+_project_region_cache: dict[str, tuple[str, float]] = {}
+
+
+def resolve_region_endpoint(base_endpoint: str, region: str | None) -> str:
+    """Return the endpoint for a project homed in ``region`` by prefixing the
+    region subdomain onto the configured endpoint. Single-region deployments
+    (which report the ``default`` region), malformed regions, and endpoints
+    already prefixed with the region pass through unchanged."""
+    if not region or region == DEFAULT_REGION or not region.isalnum():
+        return base_endpoint
+    split = urlsplit(base_endpoint)
+    hostname = split.hostname or ""
+    if not hostname or hostname.startswith(f"{region}."):
+        return base_endpoint
+    if "@" in split.netloc:
+        # Prefixing the netloc would land the region on the userinfo, not the
+        # host; credential-bearing endpoints are never regional, so pass through.
+        return base_endpoint
+    return urlunsplit(split._replace(netloc=f"{region}.{split.netloc}"))
+
+
+def _lookup_project_region(
+    console_project_id: str, bearer_token: str, target_project: str
+) -> str | None:
+    """Fetch ``target_project``'s home region from the console API. Project
+    metadata is global, so this succeeds from any gateway. Successful lookups
+    are cached; on failure the caller falls back to the configured endpoint,
+    preserving the previous behavior."""
+    cached = _project_region_cache.get(target_project)
+    if cached and cached[1] > time.monotonic():
+        return cached[0]
+    client = build_client_for_request(console_project_id, bearer_token)
+    try:
+        project = client.call(
+            "get",
+            f"/projects/{target_project}",
+            # The SDK does not turn set_project into a header on raw call();
+            # send the console project header explicitly (as context.py does).
+            headers={
+                "accept": "application/json",
+                "x-appwrite-project": console_project_id,
+            },
+            params={},
+        )
+    except Exception:
+        return None
+    region = project.get("region") if isinstance(project, dict) else None
+    if not isinstance(region, str) or not region:
+        return None
+    _project_region_cache[target_project] = (
+        region,
+        time.monotonic() + CACHE_TTL_SECONDS,
+    )
+    return region
+
+
 def resolve_client(
     target_project: str | None = None, organization_id: str | None = None
 ) -> Client:
     """Build the Appwrite client for the current request from its OAuth access
     token. The token is read from the request context populated by the auth
     middleware and carries the project it was issued for (the console). Pass
     ``target_project``/``organization_id`` to scope the call to one of the user's
-    own projects/organizations."""
+    own projects/organizations. Project-scoped calls are routed to the target
+    project's home-region endpoint (see ``resolve_region_endpoint``)."""
     access_token = get_access_token()
     if access_token is None:
         raise RuntimeError("No authenticated Appwrite access token in request context.")
@@ -237,9 +304,16 @@ def resolve_client(
     project_id = claims.get("project_id")
     if not project_id:
         raise RuntimeError("Authenticated token is missing a project identifier.")
+
+    base_endpoint = os.getenv("APPWRITE_ENDPOINT", DEFAULT_ENDPOINT)
+    endpoint = base_endpoint
+    if target_project:
+        region = _lookup_project_region(project_id, access_token.token, target_project)
+        endpoint = resolve_region_endpoint(base_endpoint, region)
     return build_client_for_request(
         project_id,
         access_token.token,
+        endpoint=endpoint,
         target_project=target_project,
         organization_id=organization_id,
     )

tests/unit/test_auth.py

@@ -127,7 +127,7 @@ def test_discovery_cache_expires_after_ttl(self):
             # Age the entry past the TTL.
             fetched_at, doc = auth._discovery_cache[pid]
             auth._discovery_cache[pid] = (
-                fetched_at - auth.DISCOVERY_TTL_SECONDS - 1,
+                fetched_at - auth.CACHE_TTL_SECONDS - 1,
                 doc,
             )
             self.assertIsNone(auth._cached_discovery(pid))
@@ -142,7 +142,7 @@ def test_stale_discovery_served_when_refresh_fails(self):
         auth._store_discovery(pid, stale_doc)
         fetched_at, doc = auth._discovery_cache[pid]
         auth._discovery_cache[pid] = (
-            fetched_at - auth.DISCOVERY_TTL_SECONDS - 1,
+            fetched_at - auth.CACHE_TTL_SECONDS - 1,
             doc,
         )
 

tests/unit/test_server.py

@@ -5,7 +5,7 @@
 import tempfile
 import unittest
 from pathlib import Path
-from unittest.mock import patch
+from unittest.mock import Mock, patch
 
 import mcp.types as types
 from appwrite.enums.browser import Browser
@@ -23,6 +23,7 @@
     build_operator,
     parse_args,
     register_services,
+    resolve_region_endpoint,
     validate_services,
 )
 from mcp_server_appwrite.tool_manager import ToolManager
@@ -617,5 +618,58 @@ def test_http_instructions_mention_url_upload(self):
         self.assertNotIn("upload", stdio.lower())
 
 
+class RegionRoutingTests(unittest.TestCase):
+    BASE = "https://cloud.appwrite.io/v1"
+
+    def setUp(self):
+        server_module._project_region_cache.clear()
+
+    def test_resolve_region_endpoint(self):
+        self.assertEqual(
+            resolve_region_endpoint(self.BASE, "sgp"),
+            "https://sgp.cloud.appwrite.io/v1",
+        )
+        # No region, single-region deployments, malformed regions, and
+        # already-prefixed endpoints pass through unchanged.
+        for region in (None, "default", "sgp.evil.example", "sgp/../x", ""):
+            self.assertEqual(resolve_region_endpoint(self.BASE, region), self.BASE)
+        prefixed = "https://sgp.cloud.appwrite.io/v1"
+        self.assertEqual(resolve_region_endpoint(prefixed, "sgp"), prefixed)
+
+    def test_lookup_project_region_caches_successful_lookups(self):
+        client = Mock()
+        client.call.return_value = {"region": "sgp"}
+        with patch.object(
+            server_module, "build_client_for_request", return_value=client
+        ):
+            for _ in range(2):
+                region = server_module._lookup_project_region("console", "tok", "proj")
+                self.assertEqual(region, "sgp")
+        client.call.assert_called_once()
+
+    def test_lookup_project_region_failure_falls_back_uncached(self):
+        client = Mock()
+        client.call.side_effect = RuntimeError("console unavailable")
+        with patch.object(
+            server_module, "build_client_for_request", return_value=client
+        ):
+            self.assertIsNone(
+                server_module._lookup_project_region("console", "tok", "proj")
+            )
+        self.assertEqual(server_module._project_region_cache, {})
+
+    def test_resolve_client_routes_target_project_to_home_region(self):
+        token = Mock(token="tok", claims={"project_id": "console"})
+        with (
+            patch.dict(os.environ, {}, clear=True),
+            patch.object(server_module, "get_access_token", return_value=token),
+            patch.object(server_module, "_lookup_project_region", return_value="sgp"),
+        ):
+            client = server_module.resolve_client(target_project="proj")
+        # _endpoint is SDK-internal, but it is the only place the resolved
+        # endpoint is observable without a network call (context.py reads it too).
+        self.assertEqual(client._endpoint, "https://sgp.cloud.appwrite.io/v1")
+
+
 if __name__ == "__main__":
     unittest.main()