feat: unify NocoBase API key authentication

tests/test_server.py

@@ -6,7 +6,7 @@ import httpx
 from starlette.responses import JSONResponse
 
 from nocobase_crm_mcp import server as server_module
-from nocobase_crm_mcp.nb_client import NbCommandError
+from nocobase_crm_mcp.nb_client import NbAuthenticationError, NbCommandError
 from nocobase_crm_mcp.server import (
     DEFAULT_HTTP_HOST,
     DEFAULT_HTTP_PATH,
@@ -15,6 +15,7 @@ from nocobase_crm_mcp.server import (
     FastMCP,
     create_http_app,
     create_server,
+    _read_nocobase_environment,
     run_http_server,
 )
 
@@ -78,6 +79,11 @@ class ErrorCrmService(FakeCrmService):
         raise NbCommandError("nb command failed with exit code 1: denied")
 
 
+class AuthenticationErrorCrmService(FakeCrmService):
+    def list_customers(self, **kwargs):
+        raise NbAuthenticationError("Request failed with status 401")
+
+
 class MissingNbCrmService(FakeCrmService):
     def crm_health(self):
         raise FileNotFoundError(2, "No such file or directory", "nb")
@@ -179,6 +185,18 @@ class ServerTests(unittest.TestCase):
         self.assertIn("denied", structured["error"]["message"])
         self.assertFalse(structured["error"]["retryable"])
 
+    def test_tools_return_specific_error_for_invalid_nocobase_api_key(self) -> None:
+        if FastMCP is None:
+            self.skipTest("MCP SDK is not installed")
+
+        server = create_server(crm_service=AuthenticationErrorCrmService(), readonly=True)
+
+        _content, structured = asyncio.run(server.call_tool("list_customers", {}))
+
+        self.assertEqual(structured["error"]["code"], "nb_auth_failed")
+        self.assertIn("NOCOBASE_MCP_API_TOKEN", structured["error"]["hint"])
+        self.assertFalse(structured["error"]["retryable"])
+
     def test_tools_return_specific_error_when_nb_is_not_found(self) -> None:
         if FastMCP is None:
             self.skipTest("MCP SDK is not installed")
@@ -230,10 +248,55 @@ class ServerTests(unittest.TestCase):
         self.assertEqual(response.status_code, 200)
         self.assertEqual(response.json(), {"uploaded": True})
 
-    def test_http_write_mode_requires_upload_api_base_url(self) -> None:
+    def test_nocobase_environment_prefers_canonical_shared_credentials(self) -> None:
+        with patch.dict(
+            "os.environ",
+            {
+                "NOCOBASE_MCP_API_BASE_URL": "https://nb.example/api/",
+                "NOCOBASE_MCP_API_TOKEN": "shared-key",
+                "NOCOBASE_MCP_UPLOAD_MAX_BYTES": "123",
+            },
+            clear=True,
+        ):
+            self.assertEqual(
+                _read_nocobase_environment(),
+                ("https://nb.example/api", "shared-key", 123),
+            )
+
+    def test_nocobase_environment_accepts_matching_legacy_credentials(self) -> None:
+        with patch.dict(
+            "os.environ",
+            {
+                "NOCOBASE_MCP_UPLOAD_API_BASE_URL": "https://legacy.example/api",
+                "NOCOBASE_MCP_UPLOAD_JWT": "legacy-key",
+            },
+            clear=True,
+        ):
+            self.assertEqual(
+                _read_nocobase_environment(),
+                ("https://legacy.example/api", "legacy-key", 26214400),
+            )
+
+    def test_nocobase_environment_rejects_incomplete_or_conflicting_credentials(self) -> None:
+        cases = (
+            {"NOCOBASE_MCP_API_BASE_URL": "https://nb.example/api"},
+            {
+                "NOCOBASE_MCP_API_BASE_URL": "https://nb.example/api",
+                "NOCOBASE_MCP_API_TOKEN": "canonical",
+                "NOCOBASE_MCP_UPLOAD_API_BASE_URL": "https://other.example/api",
+                "NOCOBASE_MCP_UPLOAD_JWT": "legacy",
+            },
+        )
+
+        for values in cases:
+            with self.subTest(values=set(values)), patch.dict("os.environ", values, clear=True):
+                with self.assertRaises(RuntimeError):
+                    _read_nocobase_environment()
+
+    def test_http_write_mode_requires_shared_api_base_url(self) -> None:
         with patch.dict("os.environ", {"NOCOBASE_MCP_TOKEN": "secret"}, clear=True):
             with patch("uvicorn.run") as uvicorn_run:
-                with self.assertRaisesRegex(RuntimeError, "NOCOBASE_MCP_UPLOAD_API_BASE_URL"):
+                with self.assertRaisesRegex(RuntimeError, "NOCOBASE_MCP_API_BASE_URL"):
                     run_http_server(
                         config_path=None,
                         host=DEFAULT_HTTP_HOST,