Validate configs on the existing-connection path; MCP policy fails closed

- Creating a tool from an existing connection validates its config too;
  the connection supplies the secrets the request leaves out.
- When the connector policies cannot be read, the MCP test and save
  routes answer 503 instead of contacting the server.

docsgpt/api/user/tools/mcp.py

@@ -158,9 +158,12 @@ def _mcp_policy_error(config: dict):
             403,
         )
     except Exception:
-        # The switch is an admin preference; when the policy table cannot be
-        # read, saving the tool (which needs the database) fails on its own.
+        # Fail closed: a server whose admin switch cannot be read is not contacted.
         current_app.logger.warning("Could not read connector policies", exc_info=True)
+        return make_response(
+            jsonify({"success": False, "error": "Could not check whether this MCP server is allowed"}),
+            503,
+        )
     return None
 
 

docsgpt/api/user/tools/routes.py

@@ -445,12 +445,14 @@ def _create_connected_tool(user, data, definition, tool_instance):
     config_requirements = tool_instance.get_config_requirements()
     public, secrets = connection_service.split_secrets(data.get("config") or {}, config_requirements)
     connection_id = data.get("connection_id")
-    if not connection_id:
-        validation_errors = _validate_config(data.get("config") or {}, config_requirements)
-        if validation_errors:
-            return make_response(
-                jsonify({"success": False, "message": "Validation failed", "errors": validation_errors}), 400,
-            )
+    # An existing connection supplies the secrets the request leaves out.
+    validation_errors = _validate_config(
+        data.get("config") or {}, config_requirements, has_existing_secrets=bool(connection_id),
+    )
+    if validation_errors:
+        return make_response(
+            jsonify({"success": False, "message": "Validation failed", "errors": validation_errors}), 400,
+        )
     try:
         with db_session() as conn:
             if connection_id:

tests/api/user/test_tools_mcp_pg.py

@@ -7,6 +7,16 @@ import pytest
 from flask import Flask
 
 
+@pytest.fixture(autouse=True)
+def _mcp_servers_allowed():
+    """The admin switch fails closed without a database; these tests are not about it.
+
+    The policy itself is covered in ``tests/connectors``.
+    """
+    with patch("docsgpt.api.user.tools.mcp._mcp_policy_error", return_value=None):
+        yield
+
+
 @pytest.fixture
 def app():
     return Flask(__name__)

tests/connectors/test_service_credentials.py

@@ -487,3 +487,16 @@ class TestMcpConnectionScope:
         config = {"transport_type": "http", "connection_id": "someone-elses-connection"}
         _sanitize_mcp_transport(config)
         assert "connection_id" not in config
+
+    def test_mcp_policy_check_fails_closed_when_unreadable(self):
+        """A server whose admin switch cannot be read is not contacted."""
+        from flask import Flask
+
+        from docsgpt.api.user.tools import mcp as mcp_routes
+
+        def _broken():
+            raise RuntimeError("database is down")
+
+        with Flask(__name__).app_context(), patch.object(mcp_routes, "db_readonly", _broken):
+            resp = mcp_routes._mcp_policy_error({"server_url": "https://mcp.linear.app/mcp"})
+        assert resp is not None and resp.status_code == 503

tests/connectors/test_tool_routes.py

@@ -90,6 +90,18 @@ class TestCreateConnectedTool:
         assert resp.status_code == 200
         assert resp.get_json()["connection_id"] == first["connection_id"]
 
+    def test_existing_connection_path_still_validates_the_config(self, app, pg_conn):
+        from docsgpt.api.user.tools.routes import CreateTool
+
+        with _db(pg_conn):
+            first = _call(app, CreateTool, _telegram()).get_json()
+            with patch("docsgpt.api.user.tools.routes._validate_config",
+                       return_value={"timeout": "Timeout must be between 1 and 300"}) as validate:
+                resp = _call(app, CreateTool, _telegram(token="", connection_id=first["connection_id"]))
+        assert resp.status_code == 400
+        # The connection supplies the secret, so a missing key is not an error.
+        assert validate.call_args.kwargs["has_existing_secrets"] is True
+
     def test_someone_elses_connection_is_not_found(self, app, pg_conn):
         from docsgpt.api.user.tools.routes import CreateTool