CockroachDB MCP Server

Instructions

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

Returns:
    Table structure with columns, indexes, and primary key.

Implementation Reference

• src/cockroachdb_mcp/tools/tables.py:196-331 (handler)

  Core implementation of the describe_table tool. Parses table name, validates schema access, queries information_schema for columns, pg_index for primary key and indexes, estimates row count, and returns structured table metadata.

  async def describe_table(table: str) -> dict[str, Any]:
      """Get detailed column information for a table.
  
      Args:
          table: Table name (schema.table or just table).
  
      Returns:
          Table structure with columns, constraints, and indexes.
      """
      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"}
  
      result: dict[str, Any] = {
          "schema": schema,
          "table": table_name,
          "full_name": f"{schema}.{table_name}",
      }
  
      try:
          # Get columns
          async with conn.cursor() as cur:
              await cur.execute(
                  """
                  SELECT
                      column_name,
                      data_type,
                      is_nullable,
                      column_default,
                      character_maximum_length,
                      numeric_precision,
                      numeric_scale
                  FROM information_schema.columns
                  WHERE table_schema = %s AND table_name = %s
                  ORDER BY ordinal_position
              """,
                  (schema, table_name),
              )
              column_rows = await cur.fetchall()
  
          if not column_rows:
              return {"status": "error", "error": f"Table '{table}' not found"}
  
          columns = []
          for row in column_rows:
              col_info: dict[str, Any] = {
                  "name": row.get("column_name"),
                  "type": row.get("data_type"),
                  "nullable": row.get("is_nullable") == "YES",
                  "default": row.get("column_default"),
              }
  
              # Add length/precision info if available
              if row.get("character_maximum_length"):
                  col_info["max_length"] = row.get("character_maximum_length")
              if row.get("numeric_precision"):
                  col_info["precision"] = row.get("numeric_precision")
                  col_info["scale"] = row.get("numeric_scale")
  
              columns.append(col_info)
  
          result["columns"] = columns
          result["column_count"] = len(columns)
  
          # Get primary key
          async with conn.cursor() as cur:
              await cur.execute(
                  """
                  SELECT a.attname as column_name
                  FROM pg_index i
                  JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey)
                  JOIN pg_class c ON c.oid = i.indrelid
                  JOIN pg_namespace n ON n.oid = c.relnamespace
                  WHERE i.indisprimary
                  AND n.nspname = %s
                  AND c.relname = %s
              """,
                  (schema, table_name),
              )
              pk_rows = await cur.fetchall()
  
          result["primary_key"] = [row.get("column_name") for row in pk_rows]
  
          # Get indexes
          async with conn.cursor() as cur:
              await cur.execute(
                  """
                  SELECT
                      i.relname as index_name,
                      ix.indisunique as is_unique,
                      ix.indisprimary as is_primary,
                      array_agg(a.attname ORDER BY array_position(ix.indkey, a.attnum)) as columns
                  FROM pg_index ix
                  JOIN pg_class i ON i.oid = ix.indexrelid
                  JOIN pg_class t ON t.oid = ix.indrelid
                  JOIN pg_namespace n ON n.oid = t.relnamespace
                  JOIN pg_attribute a ON a.attrelid = t.oid AND a.attnum = ANY(ix.indkey)
                  WHERE n.nspname = %s AND t.relname = %s
                  GROUP BY i.relname, ix.indisunique, ix.indisprimary
              """,
                  (schema, table_name),
              )
              index_rows = await cur.fetchall()
  
          indexes = []
          for row in index_rows:
              indexes.append(
                  {
                      "name": row.get("index_name"),
                      "columns": row.get("columns", []),
                      "is_unique": row.get("is_unique", False),
                      "is_primary": row.get("is_primary", False),
                  }
              )
  
          result["indexes"] = indexes
  
          # Get row count estimate
          async with conn.cursor() as cur:
              await cur.execute(f"SELECT COUNT(*) as count FROM {schema}.{table_name}")
              count_row = await cur.fetchone()
              result["row_count"] = count_row["count"] if count_row else 0
  
          return result
  
      except Exception as e:
          return {"status": "error", "error": str(e)}

• src/cockroachdb_mcp/server.py:173-186 (registration)

  Registration of the describe_table tool via @mcp.tool() decorator. This wrapper function handles exceptions and delegates to the core implementation in tools.tables.

  @mcp.tool()
  async def describe_table(table: str) -> dict[str, Any]:
      """Get detailed column information for a table.
  
      Args:
          table: Table name (schema.table or just table for public schema).
  
      Returns:
          Table structure with columns, indexes, and primary key.
      """
      try:
          return await tables.describe_table(table)
      except Exception as e:
          return {"status": "error", "error": str(e)}