CockroachDB MCP Server

Instructions

Implementation Reference

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

  Core handler function implementing describe_table tool logic: parses table name, queries information_schema for columns, pg_index for primary keys and indexes, estimates row count.
  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)

  MCP tool registration for describe_table using FastMCP @mcp.tool() decorator, delegating to tables.describe_table implementation.
  @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)}