fix: harden self-hosted stdio startup validation

docs/self-hosted.md

@@ -20,6 +20,22 @@ command. The server validates the endpoint, project ID, API key, and at least on
 supported service at startup, and fails before accepting tool calls if anything is
 wrong.
 
+If your MCP client reports a reconnect failure such as Cursor `-32000`, the stdio
+process exited before the handshake — usually bad credentials, a wrong endpoint, or
+an API key missing every readable scope the startup probe tries (`tables_db`,
+`users`, `teams`, …). Run the same command in a terminal to see the real error:
+
+```bash
+APPWRITE_PROJECT_ID=<YOUR_PROJECT_ID> \
+APPWRITE_API_KEY=<YOUR_API_KEY> \
+APPWRITE_ENDPOINT=http://localhost:9501/v1 \
+uvx mcp-server-appwrite
+```
+
+For a local Docker Appwrite instance the endpoint is typically
+`http://localhost:9501/v1` (HTTP, not HTTPS). The value must be the API base
+(`…/v1`), not the Console URL.
+
 ## Connect your client
 
 <details open>

src/mcp_server_appwrite/__main__.py

@@ -1,4 +1,4 @@
 from mcp_server_appwrite import main
 
 if __name__ == "__main__":
-    main()
+    raise SystemExit(main())

src/mcp_server_appwrite/server.py

@@ -157,12 +157,26 @@ def load_environment() -> None:
         load_dotenv(dotenv_path=discovered_dotenv)
 
 
+def _normalize_endpoint(endpoint: str) -> str:
+    """Normalize Appwrite API base URLs used by the stdio transport.
+
+    Trailing slashes break some SDK path joins. A host with no ``/v1`` suffix is
+    a common self-hosted misconfig (Console URL pasted instead of the API base),
+    so append ``/v1`` when the URL has no path.
+    """
+    cleaned = endpoint.strip().rstrip("/")
+    split = urlsplit(cleaned)
+    if split.scheme and split.netloc and split.path in ("", "/"):
+        return f"{cleaned}/v1"
+    return cleaned
+
+
 def load_appwrite_config() -> AppwriteConfig:
     load_environment()
 
     project_id = os.getenv("APPWRITE_PROJECT_ID")
     api_key = os.getenv("APPWRITE_API_KEY")
-    endpoint = os.getenv("APPWRITE_ENDPOINT", DEFAULT_ENDPOINT)
+    endpoint = _normalize_endpoint(os.getenv("APPWRITE_ENDPOINT", DEFAULT_ENDPOINT))
 
     if not project_id or not api_key:
         raise ValueError(
@@ -361,41 +375,59 @@ def _validate_service(service: Service) -> None:
 
 
 def validate_services(tools_manager: ToolManager) -> None:
+    """Probe Appwrite until one configured service succeeds.
+
+    Tries ``VALIDATION_SERVICE_ORDER`` in order so an API key that lacks
+    ``tables_db``/databases scopes can still pass via ``users``, ``teams``,
+    etc. Failing the first probe alone used to kill the stdio process before
+    the MCP handshake, which MCP clients surface as opaque reconnect errors
+    (for example Cursor ``-32000``).
+    """
     if not tools_manager.services:
         return
 
     services_by_name = {
         service.service_name: service for service in tools_manager.services
     }
-    service = next(
-        (
-            services_by_name[service_name]
-            for service_name in VALIDATION_SERVICE_ORDER
-            if service_name in services_by_name
-        ),
-        None,
-    )
-    if service is None:
-        return
+    probe_errors: list[str] = []
 
-    _log_startup(f"Validating startup access via {service.service_name}")
+    for service_name in VALIDATION_SERVICE_ORDER:
+        service = services_by_name.get(service_name)
+        if service is None:
+            continue
 
-    try:
-        _validate_service(service)
-    except AppwriteException as exc:
-        raise RuntimeError(
-            "Appwrite startup validation failed during the minimal startup probe. "
-            "Check your endpoint, project ID, API key, and required scopes.\n"
-            f"- {service.service_name}: {_format_appwrite_error(exc)}"
-        ) from exc
-    except Exception as exc:
-        raise RuntimeError(
-            "Appwrite startup validation failed during the minimal startup probe. "
-            "Check your endpoint, project ID, API key, and required scopes.\n"
-            f"- {service.service_name}: {exc}"
-        ) from exc
+        _log_startup(f"Validating startup access via {service.service_name}")
+        try:
+            _validate_service(service)
+        except AppwriteException as exc:
+            probe_errors.append(
+                f"- {service.service_name}: {_format_appwrite_error(exc)}"
+            )
+            _log_startup(
+                f"Startup probe via {service.service_name} failed; trying next service"
+            )
+            continue
+        except Exception as exc:
+            probe_errors.append(f"- {service.service_name}: {exc}")
+            _log_startup(
+                f"Startup probe via {service.service_name} failed; trying next service"
+            )
+            continue
 
-    _log_startup(f"Validated startup access via {service.service_name}")
+        _log_startup(f"Validated startup access via {service.service_name}")
+        return
+
+    if not probe_errors:
+        return
+
+    raise RuntimeError(
+        "Appwrite startup validation failed during the minimal startup probe. "
+        "Check your endpoint, project ID, API key, and required scopes. "
+        "The API key needs read access to at least one of: "
+        + ", ".join(VALIDATION_SERVICE_ORDER)
+        + ".\n"
+        + "\n".join(probe_errors)
+    )
 
 
 def _unwrap_optional_type(py_type: Any) -> Any:
@@ -1390,13 +1422,20 @@ async def run_stdio() -> None:
         )
 
 
-def main():
+def main() -> int:
     """Entry point: stdio by default, or Streamable HTTP when requested."""
     load_environment()
     args = parse_args()
 
     if args.transport == "stdio":
-        asyncio.run(run_stdio())
+        try:
+            asyncio.run(run_stdio())
+        except (ValueError, RuntimeError) as exc:
+            # Config/validation failures must stay on stderr as a short message.
+            # An uncaught traceback still exits non-zero, but MCP clients only
+            # report a generic reconnect error (e.g. Cursor -32000).
+            print(f"[appwrite-mcp] ERROR: {exc}", file=sys.stderr, flush=True)
+            return 1
         return 0
 
     # Testing flags are read from the environment at request time; fold any
@@ -1410,4 +1449,4 @@ def main():
 
 
 if __name__ == "__main__":
-    main()
+    raise SystemExit(main())

tests/unit/test_server.py

@@ -22,6 +22,7 @@
     _format_appwrite_error,
     _format_tool_result,
     _mcp_request_context,
+    _normalize_endpoint,
     _prepare_arguments,
     _validate_service,
     build_client,
@@ -31,6 +32,7 @@
     build_mcp_server,
     build_operator,
     execute_registered_tool,
+    load_appwrite_config,
     parse_args,
     register_services,
     resolve_region_endpoint,
@@ -87,6 +89,49 @@ def test_parse_args_defaults_to_stdio(self):
         self.assertEqual(args.host, "0.0.0.0")
         self.assertEqual(args.port, 8000)
 
+    def test_normalize_endpoint_strips_slash_and_appends_v1(self):
+        self.assertEqual(
+            _normalize_endpoint("http://localhost:9501/"),
+            "http://localhost:9501/v1",
+        )
+        self.assertEqual(
+            _normalize_endpoint("https://appwrite.example.com"),
+            "https://appwrite.example.com/v1",
+        )
+        self.assertEqual(
+            _normalize_endpoint("https://appwrite.example.com/v1/"),
+            "https://appwrite.example.com/v1",
+        )
+
+    def test_load_appwrite_config_normalizes_endpoint(self):
+        with patch.dict(
+            os.environ,
+            {
+                "APPWRITE_PROJECT_ID": "proj",
+                "APPWRITE_API_KEY": "key",
+                "APPWRITE_ENDPOINT": "http://localhost:9501",
+            },
+            clear=True,
+        ):
+            config = load_appwrite_config()
+
+        self.assertEqual(config.endpoint, "http://localhost:9501/v1")
+
+    def test_main_stdio_prints_clean_error_on_validation_failure(self):
+        async def boom():
+            raise RuntimeError("bad credentials")
+
+        args = parse_args(["--transport", "stdio"])
+        with (
+            patch("mcp_server_appwrite.server.parse_args", return_value=args),
+            patch("mcp_server_appwrite.server.run_stdio", side_effect=boom),
+            patch("sys.stderr", new_callable=io.StringIO) as stderr,
+        ):
+            code = server_module.main()
+
+        self.assertEqual(code, 1)
+        self.assertIn("[appwrite-mcp] ERROR: bad credentials", stderr.getvalue())
+
     def test_parse_args_accepts_env_transport(self):
         with patch.dict(os.environ, {"MCP_TRANSPORT": "http", "PORT": "9000"}):
             args = parse_args([])
@@ -457,7 +502,7 @@ def list(self):
             )()
         ]
 
-        with self.assertRaisesRegex(RuntimeError, "tables_db: boom"):
+        with self.assertRaisesRegex(RuntimeError, r"tables_db: boom"):
             validate_services(manager)
 
     def test_validate_services_accepts_successful_probe(self):
@@ -506,6 +551,58 @@ def list(self):
 
         validate_services(manager)
 
+    def test_validate_services_falls_through_to_next_service(self):
+        calls = []
+
+        class FailingService:
+            def list(self):
+                calls.append("tables_db")
+                raise Exception("missing databases scope")
+
+        class SuccessfulService:
+            def list(self):
+                calls.append("users")
+                return {"total": 0}
+
+        manager = ToolManager()
+        manager.services = [
+            type(
+                "StubService",
+                (),
+                {"service_name": "tables_db", "service": FailingService()},
+            )(),
+            type(
+                "StubService",
+                (),
+                {"service_name": "users", "service": SuccessfulService()},
+            )(),
+        ]
+
+        validate_services(manager)
+        self.assertEqual(calls, ["tables_db", "users"])
+
+    def test_validate_services_aggregates_failures_across_services(self):
+        class FailingService:
+            def list(self):
+                raise Exception("boom")
+
+        manager = ToolManager()
+        manager.services = [
+            type(
+                "StubService",
+                (),
+                {"service_name": "tables_db", "service": FailingService()},
+            )(),
+            type(
+                "StubService",
+                (),
+                {"service_name": "users", "service": FailingService()},
+            )(),
+        ]
+
+        with self.assertRaisesRegex(RuntimeError, r"tables_db: boom[\s\S]*users: boom"):
+            validate_services(manager)
+
     def test_build_operator_uses_explicit_stdio_client(self):
         tool = types.Tool(
             name="users_list",