Enforce Scoped Uniqueness using UUID for Prompts, Resources, and A2A Agents; Update Endpoints to Use Unique IDs

mcpgateway/alembic/versions/e5a59c16e041_unique_const_changes_for_prompt_and_.py

@@ -0,0 +1,183 @@
+# -*- coding: utf-8 -*-
+"""unique const changes for prompt and resource
+
+Revision ID: e5a59c16e041
+Revises: 8a2934be50c0
+Create Date: 2025-10-15 11:20:53.888488
+
+"""
+
+# Standard
+from typing import Sequence, Union
+
+# Third-Party
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision: str = "e5a59c16e041"
+down_revision: Union[str, Sequence[str], None] = "8a2934be50c0"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """
+    Apply schema changes to add or update unique constraints for prompts, resources and a2a agents.
+    This migration recreates tables with updated unique constraints and preserves data.
+    Compatible with SQLite, MySQL, and PostgreSQL.
+    """
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+
+    # ### commands auto generated by Alembic - please adjust! ###
+    for tbl, constraints in {
+        "prompts": [("name", "uq_team_owner_name_prompts")],
+        "resources": [("uri", "uq_team_owner_uri_resources")],
+        "a2a_agents": [("slug", "uq_team_owner_slug_a2a_agents")],
+    }.items():
+        try:
+            print(f"Processing {tbl} for unique constraint update...")
+
+            # Get table metadata using SQLAlchemy
+            metadata = sa.MetaData()
+            table = sa.Table(tbl, metadata, autoload_with=bind)
+
+            # Create temporary table name
+            tmp_table = f"{tbl}_tmp_nounique"
+
+            # Drop temp table if it exists
+            if inspector.has_table(tmp_table):
+                op.drop_table(tmp_table)
+
+            # Create new table structure with same columns but no old unique constraints
+            new_table = sa.Table(tmp_table, metadata)
+
+            for column in table.columns:
+                # Copy column with same properties
+                new_column = column.copy()
+                new_table.append_column(new_column)
+
+            # Copy foreign key constraints
+            for fk in table.foreign_keys:
+                new_table.append_constraint(fk.constraint.copy())
+            uqs_to_copy = []
+            # # # Copy unique constraints that we're not replacing, and skip any unique constraint only on 'name'
+            if tbl == "prompts":
+                uqs_to_copy = []
+                for uq in table.constraints:
+                    if isinstance(uq, sa.UniqueConstraint) and set([col.name for col in uq.columns]) != {"name"} and not any(uq.name == c[1] if uq.name else False for c in constraints):
+                        uqs_to_copy.append(uq)
+            # Copy unique constraints that we're not replacing, and skip any unique constraint only on 'name'
+            if tbl == "resources":
+                uqs_to_copy = [
+                    uq
+                    for uq in table.constraints
+                    if isinstance(uq, sa.UniqueConstraint) and set([col.name for col in uq.columns]) != {"uri"} and not any(uq.name == c[1] if uq.name else False for c in constraints)
+                ]
+
+            # For a2a_agents, also drop any unique constraint on just 'name'
+            if tbl == "a2a_agents":
+                uqs_to_copy = [
+                    uq
+                    for uq in table.constraints
+                    if isinstance(uq, sa.UniqueConstraint)
+                    and set([col.name for col in uq.columns]) != {"name"}
+                    and set([col.name for col in uq.columns]) != {"slug"}
+                    and not any(uq.name == c[1] if uq.name else False for c in constraints)
+                ]
+            for uq in uqs_to_copy:
+                if uq is not None:
+                    new_table.append_constraint(uq.copy())
+
+            # Create the temporary table
+            new_table.create(bind)
+
+            # Copy data
+            column_names = [c.name for c in table.columns]
+            insert_stmt = new_table.insert().from_select(column_names, sa.select(*[table.c[name] for name in column_names]))
+            bind.execute(insert_stmt)
+
+            # Add new unique constraints using batch operations for SQLite compatibility
+            with op.batch_alter_table(tmp_table, schema=None) as batch_op:
+                for col, constraint_name in constraints:
+                    cols = ["team_id", "owner_email", col]
+                    batch_op.create_unique_constraint(constraint_name, cols)
+
+            # Drop original table and rename temp table
+            op.drop_table(tbl)
+            op.rename_table(tmp_table, tbl)
+
+        except Exception as e:
+            print(f"Warning: Could not update unique constraint on {tbl} table: {e}")
+        # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """
+    Revert schema changes, restoring previous unique constraints for prompts, resources and a2a_agents.
+    This migration recreates tables with the original unique constraints and preserves data.
+    Compatible with SQLite, MySQL, and PostgreSQL.
+    """
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+
+    for tbl, constraints in {
+        "prompts": [("name", "uq_team_owner_name_prompts")],
+        "resources": [("uri", "uq_team_owner_uri_resources")],
+        "a2a_agents": [("slug", "uq_team_owner_slug_a2a_agents")],
+    }.items():
+        try:
+            print(f"Processing {tbl} for unique constraint revert...")
+
+            # Get table metadata using SQLAlchemy
+            metadata = sa.MetaData()
+            table = sa.Table(tbl, metadata, autoload_with=bind)
+
+            # Create temporary table name
+            tmp_table = f"{tbl}_tmp_revert"
+
+            # Drop temp table if it exists
+            if inspector.has_table(tmp_table):
+                op.drop_table(tmp_table)
+
+            # Create new table structure with same columns but original unique constraints
+            new_table = sa.Table(tmp_table, metadata)
+
+            for column in table.columns:
+                # Copy column with same properties
+                new_column = column.copy()
+                new_table.append_column(new_column)
+
+            # Copy foreign key constraints
+            for fk in table.foreign_keys:
+                new_table.append_constraint(fk.constraint.copy())
+
+            # Copy unique constraints that we're not reverting
+            uqs_to_copy = [uq for uq in table.constraints if isinstance(uq, sa.UniqueConstraint) and not any(uq.name == c[1] if uq.name else False for c in constraints)]
+            for uq in uqs_to_copy:
+                new_table.append_constraint(uq.copy())
+
+            # Add back the original single-column unique constraints
+
+            for col, _ in constraints:
+                if col in [c.name for c in table.columns]:
+                    new_table.append_constraint(sa.UniqueConstraint(col))
+            if tbl == "a2a_agents":
+                # Also re-add unique constraint on 'name' for a2a_agents
+                new_table.append_constraint(sa.UniqueConstraint("name"))
+            # Create the temporary table
+            new_table.create(bind)
+
+            # Copy data
+            column_names = [c.name for c in table.columns]
+            insert_stmt = new_table.insert().from_select(column_names, sa.select(*[table.c[name] for name in column_names]))
+            bind.execute(insert_stmt)
+
+            # Drop original table and rename temp table
+            op.drop_table(tbl)
+            op.rename_table(tmp_table, tbl)
+
+        except Exception as e:
+            print(f"Warning: Could not revert unique constraint on {tbl} table: {e}")
+    # ### end Alembic commands ###

mcpgateway/db.py

@@ -1840,7 +1840,7 @@ class Resource(Base):
     __tablename__ = "resources"
 
     id: Mapped[int] = mapped_column(primary_key=True)
-    uri: Mapped[str] = mapped_column(String(767), unique=True)
+    uri: Mapped[str] = mapped_column(String(767), nullable=False)
     name: Mapped[str] = mapped_column(String(255), nullable=False)
     description: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
     mime_type: Mapped[Optional[str]] = mapped_column(String(255), nullable=True)
@@ -1881,6 +1881,7 @@ class Resource(Base):
 
     # Many-to-many relationship with Servers
     servers: Mapped[List["Server"]] = relationship("Server", secondary=server_resource_association, back_populates="resources")
+    __table_args__ = (UniqueConstraint("team_id", "owner_email", "uri", name="uq_team_owner_uri_resource"),)
 
     @property
     def content(self) -> "ResourceContent":
@@ -1927,13 +1928,15 @@ def content(self) -> "ResourceContent":
         if self.text_content is not None:
             return ResourceContent(
                 type="resource",
+                id=str(self.id),
                 uri=self.uri,
                 mime_type=self.mime_type,
                 text=self.text_content,
             )
         if self.binary_content is not None:
             return ResourceContent(
                 type="resource",
+                id=str(self.id),
                 uri=self.uri,
                 mime_type=self.mime_type or "application/octet-stream",
                 blob=self.binary_content,
@@ -2078,7 +2081,7 @@ class Prompt(Base):
     __tablename__ = "prompts"
 
     id: Mapped[int] = mapped_column(primary_key=True)
-    name: Mapped[str] = mapped_column(String(255), unique=True)
+    name: Mapped[str] = mapped_column(String(255), nullable=False)
     description: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
     template: Mapped[str] = mapped_column(Text)
     argument_schema: Mapped[Dict[str, Any]] = mapped_column(JSON)
@@ -2116,6 +2119,8 @@ class Prompt(Base):
     owner_email: Mapped[Optional[str]] = mapped_column(String(255), nullable=True)
     visibility: Mapped[str] = mapped_column(String(20), nullable=False, default="public")
 
+    __table_args__ = (UniqueConstraint("team_id", "owner_email", "name", name="uq_team_owner_name_prompt"),)
+
     def validate_arguments(self, args: Dict[str, str]) -> None:
         """
         Validate prompt arguments against the argument schema.
@@ -2548,8 +2553,8 @@ class A2AAgent(Base):
     __tablename__ = "a2a_agents"
 
     id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: uuid.uuid4().hex)
-    name: Mapped[str] = mapped_column(String(255), nullable=False, unique=True)
-    slug: Mapped[str] = mapped_column(String(255), nullable=False, unique=True)
+    name: Mapped[str] = mapped_column(String(255), nullable=False)
+    slug: Mapped[str] = mapped_column(String(255), nullable=False)
     description: Mapped[Optional[str]] = mapped_column(Text)
     endpoint_url: Mapped[str] = mapped_column(String(767), nullable=False)
     agent_type: Mapped[str] = mapped_column(String(50), nullable=False, default="generic")  # e.g., "openai", "anthropic", "custom"
@@ -2594,6 +2599,7 @@ class A2AAgent(Base):
     # Relationships
     servers: Mapped[List["Server"]] = relationship("Server", secondary=server_a2a_association, back_populates="a2a_agents")
     metrics: Mapped[List["A2AAgentMetric"]] = relationship("A2AAgentMetric", back_populates="a2a_agent", cascade="all, delete-orphan")
+    __table_args__ = (UniqueConstraint("team_id", "owner_email", "slug", name="uq_team_owner_slug_a2a_agent"),)
 
     @property
     def execution_count(self) -> int: