CockroachDB MCP Server

Instructions

Args:
    table: Table name (schema.table or just table).

Returns:
    Table statistics including row count and size.

Implementation Reference

• src/cockroachdb_mcp/tools/tables.py:334-384 (handler)

  Core handler function that executes the tool logic: parses table name, checks schema permissions, queries row count and table size using CockroachDB pg_total_relation_size.

  async def get_table_stats(table: str) -> dict[str, Any]:
      """Get statistics for a table.
  
      Args:
          table: Table name (schema.table or just table).
  
      Returns:
          Table statistics.
      """
      conn = await connection_manager.ensure_connected()
  
      # Parse schema and table name
      if "." in table:
          schema, table_name = table.rsplit(".", 1)
      else:
          schema = "public"
          table_name = table
  
      # Check if schema is allowed
      if not _is_allowed_schema(schema):
          return {"status": "error", "error": f"Schema '{schema}' is not allowed"}
  
      try:
          # Get table size and row count
          async with conn.cursor() as cur:
              # Row count
              await cur.execute(f"SELECT COUNT(*) as count FROM {schema}.{table_name}")
              count_row = await cur.fetchone()
              row_count = count_row["count"] if count_row else 0
  
              # Table size using CockroachDB specific query
              await cur.execute(
                  """
                  SELECT
                      pg_size_pretty(pg_total_relation_size(%s::regclass)) as total_size,
                      pg_total_relation_size(%s::regclass) as total_bytes
              """,
                  (f"{schema}.{table_name}", f"{schema}.{table_name}"),
              )
              size_row = await cur.fetchone()
  
          return {
              "schema": schema,
              "table": table_name,
              "full_name": f"{schema}.{table_name}",
              "row_count": row_count,
              "total_size": size_row.get("total_size") if size_row else None,
              "total_bytes": size_row.get("total_bytes") if size_row else None,
          }
      except Exception as e:
          return {"status": "error", "error": str(e)}

• src/cockroachdb_mcp/server.py:189-202 (registration)

  MCP tool registration via @mcp.tool() decorator. Thin wrapper that delegates to the core implementation in tables.get_table_stats and handles exceptions.

  @mcp.tool()
  async def get_table_stats(table: str) -> dict[str, Any]:
      """Get statistics for a table.
  
      Args:
          table: Table name (schema.table or just table).
  
      Returns:
          Table statistics including row count and size.
      """
      try:
          return await tables.get_table_stats(table)
      except Exception as e:
          return {"status": "error", "error": str(e)}