Improved gateway duplication check allowing same url with certain conditions

mcpgateway/admin.py

@@ -102,7 +102,7 @@
 from mcpgateway.services.catalog_service import catalog_service
 from mcpgateway.services.encryption_service import get_encryption_service
 from mcpgateway.services.export_service import ExportError, ExportService
-from mcpgateway.services.gateway_service import GatewayConnectionError, GatewayNameConflictError, GatewayNotFoundError, GatewayService, GatewayUrlConflictError
+from mcpgateway.services.gateway_service import GatewayConnectionError, GatewayDuplicateConflictError, GatewayNameConflictError, GatewayNotFoundError, GatewayService
 from mcpgateway.services.import_service import ConflictStrategy
 from mcpgateway.services.import_service import ImportError as ImportServiceError
 from mcpgateway.services.import_service import ImportService, ImportValidationError
@@ -6846,7 +6846,7 @@ async def admin_add_gateway(request: Request, db: Session = Depends(get_db), use
 
     except GatewayConnectionError as ex:
         return JSONResponse(content={"message": str(ex), "success": False}, status_code=502)
-    except GatewayUrlConflictError as ex:
+    except GatewayDuplicateConflictError as ex:
         return JSONResponse(content={"message": str(ex), "success": False}, status_code=409)
     except GatewayNameConflictError as ex:
         return JSONResponse(content={"message": str(ex), "success": False}, status_code=409)

mcpgateway/alembic/versions/f3a3a3d901b8_remove_gateway_url_unique_constraint.py

@@ -0,0 +1,98 @@
+# -*- coding: utf-8 -*-
+"""Location: ./mcpgateway/alembic/versions/f3a3a3d901b8_remove_gateway_url_unique_constraint.py
+Copyright 2025
+SPDX-License-Identifier: Apache-2.0
+Authors: Keval Mahajan
+
+Alembic migration to remove unique constraint on gateway URL.
+An improved alternative duplication check has been implemented for gateway duplication prevention.
+
+Revision ID: f3a3a3d901b8
+Revises: aac21d6f9522
+Create Date: 2025-11-11 22:30:05.474282
+
+"""
+
+# Standard
+from typing import Sequence, Union
+
+# Third-Party
+from alembic import op
+from sqlalchemy.engine import Inspector
+
+# revision identifiers, used by Alembic.
+revision: str = "f3a3a3d901b8"
+down_revision: Union[str, Sequence[str], None] = "aac21d6f9522"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def constraint_exists(inspector, table_name, constraint_name):
+    """
+    Check if a specific unique constraint exists on a given table.
+
+    This function queries the database using the provided SQLAlchemy
+    inspector to determine if a constraint with the given name exists.
+    If the check fails due to an exception (e.g., database connectivity issues),
+    it conservatively assumes that the constraint exists.
+
+    Args:
+        inspector (sqlalchemy.engine.reflection.Inspector): SQLAlchemy inspector
+            instance for database introspection.
+        table_name (str): Name of the table to inspect.
+        constraint_name (str): Name of the unique constraint to check.
+
+    Returns:
+        bool: True if the constraint exists or if the check could not be performed,
+              False if the constraint does not exist.
+    """
+    try:
+        unique_constraints = inspector.get_unique_constraints(table_name)
+        return any(uc["name"] == constraint_name for uc in unique_constraints)
+    except Exception:
+        # Fallback: assume constraint exists if we can't check
+        return True
+
+
+def upgrade():
+    """Remove the unique constraint on (team_id, owner_email, url) from gateway table."""
+
+    conn = op.get_bind()
+    inspector = Inspector.from_engine(conn)
+
+    # Check if constraint exists before attempting to drop
+    if not constraint_exists(inspector, "gateways", "uq_team_owner_url_gateway"):
+        print("Constraint 'uq_team_owner_url_gateway' does not exist, skipping drop.")
+        return
+
+    if conn.dialect.name == "sqlite":
+        # SQLite: Use batch mode to recreate table without the constraint
+        with op.batch_alter_table("gateways", schema=None) as batch_op:
+            batch_op.drop_constraint("uq_team_owner_url_gateway", type_="unique")
+    else:
+        # PostgreSQL, MySQL, etc.: Direct constraint drop
+        op.drop_constraint("uq_team_owner_url_gateway", "gateways", type_="unique")
+
+    print("Successfully removed constraint 'uq_team_owner_url_gateway' from gateway table.")
+
+
+def downgrade():
+    """Re-add the unique constraint on (team_id, owner_email, url) to gateway table."""
+
+    conn = op.get_bind()
+    inspector = Inspector.from_engine(conn)
+
+    # Check if constraint already exists before attempting to create
+    if constraint_exists(inspector, "gateways", "uq_team_owner_url_gateway"):
+        print("Constraint 'uq_team_owner_url_gateway' already exists, skipping creation.")
+        return
+
+    if conn.dialect.name == "sqlite":
+        # SQLite: Use batch mode to recreate table with the constraint
+        with op.batch_alter_table("gateways", schema=None) as batch_op:
+            batch_op.create_unique_constraint("uq_team_owner_url_gateway", ["team_id", "owner_email", "url"])
+    else:
+        # PostgreSQL, MySQL, etc.: Direct constraint creation
+        op.create_unique_constraint("uq_team_owner_url_constraint", "gateways", ["team_id", "owner_email", "url"])
+
+    print("Successfully re-added constraint 'uq_team_owner_url_gateway' to gateways table.")

mcpgateway/db.py

@@ -2818,10 +2818,7 @@ class Gateway(Base):
 
     registered_oauth_clients: Mapped[List["RegisteredOAuthClient"]] = relationship("RegisteredOAuthClient", back_populates="gateway", cascade="all, delete-orphan")
 
-    __table_args__ = (
-        UniqueConstraint("team_id", "owner_email", "slug", name="uq_team_owner_slug_gateway"),
-        UniqueConstraint("team_id", "owner_email", "url", name="uq_team_owner_url_gateway"),
-    )
+    __table_args__ = (UniqueConstraint("team_id", "owner_email", "slug", name="uq_team_owner_slug_gateway"),)
 
 
 @event.listens_for(Gateway, "after_update")

mcpgateway/main.py

@@ -108,7 +108,7 @@
 from mcpgateway.services.a2a_service import A2AAgentError, A2AAgentNameConflictError, A2AAgentNotFoundError, A2AAgentService
 from mcpgateway.services.completion_service import CompletionService
 from mcpgateway.services.export_service import ExportError, ExportService
-from mcpgateway.services.gateway_service import GatewayConnectionError, GatewayError, GatewayNameConflictError, GatewayNotFoundError, GatewayService, GatewayUrlConflictError
+from mcpgateway.services.gateway_service import GatewayConnectionError, GatewayDuplicateConflictError, GatewayError, GatewayNameConflictError, GatewayNotFoundError, GatewayService
 from mcpgateway.services.import_service import ConflictStrategy, ImportConflictError
 from mcpgateway.services.import_service import ImportError as ImportServiceError
 from mcpgateway.services.import_service import ImportService, ImportValidationError
@@ -3415,8 +3415,8 @@ async def register_gateway(
             return JSONResponse(content={"message": "Unable to process input"}, status_code=status.HTTP_400_BAD_REQUEST)
         if isinstance(ex, GatewayNameConflictError):
             return JSONResponse(content={"message": "Gateway name already exists"}, status_code=status.HTTP_409_CONFLICT)
-        if isinstance(ex, GatewayUrlConflictError):
-            return JSONResponse(content={"message": "Gateway URL already exists"}, status_code=status.HTTP_409_CONFLICT)
+        if isinstance(ex, GatewayDuplicateConflictError):
+            return JSONResponse(content={"message": "Gateway already exists"}, status_code=status.HTTP_409_CONFLICT)
         if isinstance(ex, RuntimeError):
             return JSONResponse(content={"message": "Error during execution"}, status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)
         if isinstance(ex, ValidationError):
@@ -3493,8 +3493,8 @@ async def update_gateway(
             return JSONResponse(content={"message": "Unable to process input"}, status_code=status.HTTP_400_BAD_REQUEST)
         if isinstance(ex, GatewayNameConflictError):
             return JSONResponse(content={"message": "Gateway name already exists"}, status_code=status.HTTP_409_CONFLICT)
-        if isinstance(ex, GatewayUrlConflictError):
-            return JSONResponse(content={"message": "Gateway URL already exists"}, status_code=status.HTTP_409_CONFLICT)
+        if isinstance(ex, GatewayDuplicateConflictError):
+            return JSONResponse(content={"message": "Gateway already exists"}, status_code=status.HTTP_409_CONFLICT)
         if isinstance(ex, RuntimeError):
             return JSONResponse(content={"message": "Error during execution"}, status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)
         if isinstance(ex, ValidationError):