feat(mcp): add entity read tools for projects, connections, table groups, test suites (TG-1053)

Adds six read tools so an MCP client can introspect TestGen configuration
before invoking write tools: get_project, list_connections, get_connection,
list_table_groups, get_table_group, get_test_suite. All gate on the view
permission; get_connection redacts every encrypted field; get_test_suite
rejects monitor suites with the unified missing-or-inaccessible wording;
sql_flavor_code is rendered as its display label everywhere.

Backing model helpers: TestSuite.test_definition_stats aggregates total /
locked counts plus per-test-type breakdown in one query; Connection
.list_for_project and TableGroup.list_for_project / list_for_connection
return paginated config-focused list rows with relevant activity timestamps.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: diogodk

testgen/common/models/connection.py

@@ -43,6 +43,17 @@ class ConnectionMinimal(EntityMinimal):
     connection_name: str
 
 
+@dataclass
+class ConnectionListItem(EntityMinimal):
+    connection_id: int
+    connection_name: str
+    project_code: str
+    sql_flavor_code: SQLFlavorCode
+    project_host: str | None
+    project_db: str | None
+    table_group_count: int
+
+
 class Connection(Entity):
     __tablename__ = "connections"
 
@@ -93,6 +104,35 @@ def select_minimal_where(
         results = cls._select_columns_where(cls._minimal_columns, *clauses, order_by=order_by)
         return [ConnectionMinimal(**row) for row in results]
 
+    @classmethod
+    def list_for_project(
+        cls, project_code: str, *clauses, page: int = 1, limit: int = 20
+    ) -> tuple[list[ConnectionListItem], int]:
+        """Paginated listing of connections in a project, with their table-group counts."""
+        tg_count_subq = (
+            select(
+                TableGroup.connection_id.label("connection_id"),
+                func.count().label("table_group_count"),
+            )
+            .group_by(TableGroup.connection_id)
+            .subquery()
+        )
+        query = (
+            select(
+                cls.connection_id,
+                cls.connection_name,
+                cls.project_code,
+                cls.sql_flavor_code,
+                cls.project_host,
+                cls.project_db,
+                func.coalesce(tg_count_subq.c.table_group_count, 0).label("table_group_count"),
+            )
+            .outerjoin(tg_count_subq, tg_count_subq.c.connection_id == cls.connection_id)
+            .where(cls.project_code == project_code, *clauses)
+            .order_by(*cls._default_order_by)
+        )
+        return cls._paginate(query, page=page, limit=limit, data_class=ConnectionListItem)
+
     @classmethod
     def is_in_use(cls, ids: list[str]) -> bool:
         table_groups = TableGroup.select_minimal_where(TableGroup.connection_id.in_(ids))

testgen/common/models/table_group.py

@@ -47,6 +47,19 @@ class TableGroupStats(EntityMinimal):
     data_point_ct: int
 
 
+@dataclass
+class TableGroupListItem(EntityMinimal):
+    id: UUID
+    table_groups_name: str
+    table_group_schema: str
+    project_code: str
+    connection_name: str | None
+    table_count: int
+    last_profiled_date: datetime | None
+    last_tested_date: datetime | None
+    testing_score: float | None
+
+
 @dataclass
 class TableGroupSummary(EntityMinimal):
     id: UUID
@@ -385,6 +398,89 @@ def select_summary(
         total = items[0].total_count if items else 0
         return items, total
 
+    @classmethod
+    def list_for_project(
+        cls, project_code: str, *, page: int = 1, limit: int = 20
+    ) -> tuple[list[TableGroupListItem], int]:
+        """Config-focused paginated listing for a project, with table count and activity timestamps."""
+        return cls._list_with_activity(
+            scope_sql="groups.project_code = :project_code",
+            scope_params={"project_code": project_code},
+            page=page,
+            limit=limit,
+        )
+
+    @classmethod
+    def list_for_connection(
+        cls, connection_id: int, *, page: int = 1, limit: int = 20
+    ) -> tuple[list[TableGroupListItem], int]:
+        """Config-focused paginated listing for a connection, with table count and activity timestamps."""
+        return cls._list_with_activity(
+            scope_sql="groups.connection_id = :connection_id",
+            scope_params={"connection_id": connection_id},
+            page=page,
+            limit=limit,
+        )
+
+    @classmethod
+    def _list_with_activity(
+        cls,
+        *,
+        scope_sql: str,
+        scope_params: dict,
+        page: int,
+        limit: int,
+    ) -> tuple[list[TableGroupListItem], int]:
+        params: dict = {**scope_params, "limit": limit, "offset": (page - 1) * limit}
+        query = f"""
+        WITH stats AS (
+            SELECT table_groups_id, COUNT(*) AS table_count
+            FROM data_table_chars
+            GROUP BY table_groups_id
+        ),
+        latest_profile AS (
+            SELECT pr.table_groups_id, MAX(je.started_at) AS started_at
+            FROM profiling_runs pr
+                LEFT JOIN job_executions je ON je.id = pr.job_execution_id
+            GROUP BY pr.table_groups_id
+        ),
+        latest_test AS (
+            SELECT ts.table_groups_id, MAX(tr.test_starttime) AS test_starttime
+            FROM test_runs tr
+                JOIN test_suites ts ON ts.id = tr.test_suite_id
+            WHERE ts.is_monitor IS NOT TRUE
+            GROUP BY ts.table_groups_id
+        )
+        SELECT
+            groups.id,
+            groups.table_groups_name,
+            groups.table_group_schema,
+            groups.project_code,
+            connections.connection_name,
+            COALESCE(stats.table_count, 0) AS table_count,
+            latest_profile.started_at AS last_profiled_date,
+            latest_test.test_starttime AS last_tested_date,
+            groups.dq_score_testing AS testing_score,
+            COUNT(*) OVER() AS total_count
+        FROM table_groups AS groups
+            LEFT JOIN connections ON connections.connection_id = groups.connection_id
+            LEFT JOIN stats ON stats.table_groups_id = groups.id
+            LEFT JOIN latest_profile ON latest_profile.table_groups_id = groups.id
+            LEFT JOIN latest_test ON latest_test.table_groups_id = groups.id
+        WHERE {scope_sql}
+        ORDER BY LOWER(groups.table_groups_name)
+        LIMIT :limit OFFSET :offset;
+        """
+        rows = get_current_session().execute(text(query), params).mappings().all()
+        if not rows:
+            return [], 0
+        total = int(rows[0]["total_count"])
+        items = [
+            TableGroupListItem(**{k: v for k, v in row.items() if k != "total_count"})
+            for row in rows
+        ]
+        return items, total
+
     @classmethod
     def is_in_use(cls, ids: list[str]) -> bool:
         test_suites = TestSuite.select_minimal_where(TestSuite.table_groups_id.in_(ids))

testgen/common/models/test_suite.py

@@ -29,6 +29,15 @@ class TestSuiteMinimal(EntityMinimal):
     export_to_observability: str
 
 
+@dataclass(frozen=True)
+class TestDefinitionStats:
+    """Aggregate test-definition counts for a test suite."""
+
+    total: int
+    locked: int
+    counts_by_type: dict[str, int]
+
+
 @dataclass
 class TestSuiteSummary(EntityMinimal):
     id: UUID
@@ -213,6 +222,30 @@ def select_summary(cls, project_code: str, table_group_id: str | UUID | None = N
         results = db_session.execute(text(query), params).mappings().all()
         return [TestSuiteSummary(**row) for row in results]
 
+    @classmethod
+    def test_definition_stats(cls, test_suite_id: str | UUID) -> "TestDefinitionStats":
+        """Aggregate test-definition counts for a suite: total, locked, and per-test-type."""
+        query = """
+        SELECT
+            test_type,
+            COUNT(*) AS total,
+            SUM(CASE WHEN COALESCE(lock_refresh, 'N') = 'Y' THEN 1 ELSE 0 END) AS locked
+        FROM test_definitions
+        WHERE test_suite_id = :test_suite_id
+        GROUP BY test_type
+        ORDER BY test_type;
+        """
+        rows = (
+            get_current_session()
+            .execute(text(query), {"test_suite_id": str(test_suite_id)})
+            .mappings()
+            .all()
+        )
+        counts_by_type = {row["test_type"]: int(row["total"]) for row in rows}
+        total = sum(counts_by_type.values())
+        locked = sum(int(row["locked"]) for row in rows)
+        return TestDefinitionStats(total=total, locked=locked, counts_by_type=counts_by_type)
+
     @classmethod
     def is_in_use(cls, ids: list[str]) -> bool:
         query = """
@@ -248,4 +281,3 @@ def cascade_delete(cls, ids: list[str]) -> None:
         db_session = get_current_session()
         db_session.execute(text(query), {"test_suite_ids": tuple(ids)})
         cls.delete_where(cls.id.in_(ids))
-

testgen/mcp/server.py

@@ -140,9 +140,20 @@ def build_mcp_server(
         profiling_overview,
         table_health,
     )
-    from testgen.mcp.tools.connections import test_connection
+    from testgen.mcp.tools.connections import (
+            get_connection,
+        list_connections,
+        test_connection,
+        )
     from testgen.mcp.tools.data_catalog import update_catalog_metadata
-    from testgen.mcp.tools.discovery import get_data_inventory, list_projects, list_tables, list_test_suites
+    from testgen.mcp.tools.discovery import (
+        get_data_inventory,
+        get_project,
+        get_test_suite,
+        list_projects,
+        list_tables,
+        list_test_suites,
+    )
     from testgen.mcp.tools.execution import (
         cancel_profiling_run,
         cancel_test_run,
@@ -208,6 +219,8 @@ def build_mcp_server(
     from testgen.mcp.tools.source_data import get_source_data, get_source_data_query, get_table_sample
     from testgen.mcp.tools.table_groups import (
         create_table_group,
+        get_table_group,
+        list_table_groups,
         preview_table_group,
         update_table_group,
     )
@@ -263,8 +276,10 @@ def safe_prompt(fn):
     # Tools
     safe_tool(get_data_inventory)
     safe_tool(list_projects)
+    safe_tool(get_project)
     safe_tool(list_tables)
     safe_tool(list_test_suites)
+    safe_tool(get_test_suite)
     safe_tool(list_test_runs)
     safe_tool(get_test_run)
     safe_tool(list_test_results)
@@ -329,7 +344,11 @@ def safe_prompt(fn):
     safe_tool(create_notification)
     safe_tool(update_notification)
     safe_tool(delete_notification)
+    safe_tool(list_connections)
+    safe_tool(get_connection)
     safe_tool(test_connection)
+    safe_tool(list_table_groups)
+    safe_tool(get_table_group)
     safe_tool(create_table_group)
     safe_tool(update_table_group)
     safe_tool(preview_table_group)

testgen/mcp/tools/connections.py

@@ -26,14 +26,115 @@
     apply_connection_params,
     connection_display_fields,
     connection_field_labels,
+    format_page_footer,
+    format_page_info,
     infer_mode,
     parse_sql_flavor,
     resolve_connection,
     validate_connection_fields,
+    validate_limit,
+    validate_page,
 )
 from testgen.mcp.tools.markdown import MdDoc
 
 
+@with_database_session
+@mcp_permission("view")
+def list_connections(project_code: str, page: int = 1, limit: int = 20) -> str:
+    """List database connections in a project with their database type and table-group counts.
+
+    Use this before changing or referencing a connection to confirm its ID, name, and host.
+    Credentials are never returned.
+
+    Args:
+        project_code: The project code to list connections for.
+        page: Page number starting at 1 (default 1).
+        limit: Page size (default 20, max 100).
+    """
+    validate_page(page)
+    validate_limit(limit, 100)
+
+    perms = get_project_permissions()
+    perms.verify_access(project_code, not_found=MCPResourceNotAccessible("Project", project_code))
+
+    rows, total = Connection.list_for_project(project_code, page=page, limit=limit)
+
+    if not rows:
+        if page > 1:
+            return f"No connections on page {page} (total: {total})."
+        return f"No connections found for project `{project_code}`."
+
+    doc = MdDoc()
+    doc.heading(1, f"Connections for `{project_code}`")
+    doc.text(format_page_info(total, page, limit))
+    table_rows: list[list[object]] = []
+    for row in rows:
+        label = SQL_FLAVOR_CODE_TO_LABEL.get(row.sql_flavor_code)
+        database_type = label.value if label else row.sql_flavor_code
+        table_rows.append(
+            [
+                row.connection_id,
+                row.connection_name,
+                database_type,
+                row.project_host,
+                row.project_db,
+                row.table_group_count,
+            ]
+        )
+    doc.table(
+        ["ID", "Name", "Type", "Host", "Database", "Table groups"],
+        table_rows,
+        code=[0, 3, 4],
+    )
+    if footer := format_page_footer(total, page, limit):
+        doc.text(footer)
+    return doc.render()
+
+
+@with_database_session
+@mcp_permission("view")
+def get_connection(connection_id: int) -> str:
+    """Get a connection's configuration, including database type, host, and authentication mode.
+
+    Credentials (password, private key, service-account key) are never returned.
+    Use this before editing a connection or creating a table group on it.
+
+    Args:
+        connection_id: Bigint connection ID returned by `list_connections` or shown on the connections page.
+    """
+    connection = resolve_connection(connection_id)
+
+    doc = MdDoc()
+    doc.heading(1, f"Connection `{connection.connection_name}`")
+    doc.field("ID", connection.connection_id, code=True)
+    doc.field("Project", connection.project_code, code=True)
+    label = SQL_FLAVOR_CODE_TO_LABEL.get(connection.sql_flavor_code)
+    doc.field("Type", label.value if label else connection.sql_flavor_code)
+
+    if connection.project_host:
+        doc.field("Host", connection.project_host, code=True)
+    if connection.project_port:
+        doc.field("Port", connection.project_port)
+    if connection.project_db:
+        doc.field("Database", connection.project_db, code=True)
+    if connection.project_user:
+        doc.field("User", connection.project_user, code=True)
+    if connection.connect_by_url and connection.url:
+        doc.field("URL", connection.url, code=True)
+    if connection.warehouse:
+        doc.field("Warehouse", connection.warehouse, code=True)
+    if connection.http_path:
+        doc.field("HTTP Path", connection.http_path, code=True)
+    doc.field("Authentication", _authentication_label(connection))
+
+    if connection.max_threads is not None:
+        doc.field("Max Threads", connection.max_threads)
+    if connection.max_query_chars is not None:
+        doc.field("Max Expression Length", connection.max_query_chars)
+
+    return doc.render()
+
+
 @with_database_session
 @mcp_permission("administer")
 def test_connection(