CockroachDB MCP Server

Instructions

Args:
    include_system: Include system databases (postgres, defaultdb, etc.).

Returns:
    List of databases.

Implementation Reference

• src/cockroachdb_mcp/tools/tables.py:24-69 (handler)

  Core handler function that executes the 'SHOW DATABASES' query, filters out blocked and system databases (unless specified), and returns a structured list of databases.

  async def list_databases(include_system: bool = False) -> dict[str, Any]:
      """List all databases in the cluster.
  
      Args:
          include_system: Include system databases.
  
      Returns:
          List of databases.
      """
      conn = await connection_manager.ensure_connected()
  
      try:
          async with conn.cursor() as cur:
              await cur.execute("SHOW DATABASES")
              rows = await cur.fetchall()
  
          databases = []
          system_dbs = {"system", "postgres", "defaultdb", "crdb_internal"}
  
          for row in rows:
              db_name = row.get("database_name", "")
  
              # Skip blocked databases
              if _is_blocked_database(db_name):
                  continue
  
              # Skip system databases unless requested
              is_system = db_name in system_dbs
              if is_system and not include_system:
                  continue
  
              databases.append(
                  {
                      "name": db_name,
                      "is_system": is_system,
                  }
              )
  
          return {
              "databases": databases,
              "count": len(databases),
              "current_database": connection_manager.current_database,
          }
      except Exception as e:
          return {"status": "error", "error": str(e)}

• src/cockroachdb_mcp/server.py:122-136 (registration)

  MCP tool registration using @mcp.tool() decorator. This wrapper function handles the tool invocation, delegates to the core implementation in tables.py, and provides error handling.

  @mcp.tool()
  async def list_databases(include_system: bool = False) -> dict[str, Any]:
      """List all databases in the cluster.
  
      Args:
          include_system: Include system databases (postgres, defaultdb, etc.).
  
      Returns:
          List of databases.
      """
      try:
          return await tables.list_databases(include_system)
      except Exception as e:
          return {"status": "error", "error": str(e)}

• src/cockroachdb_mcp/tools/tables.py:11-14 (helper)

  Helper function used by list_databases to filter out blocked databases based on configuration.

  def _is_blocked_database(db_name: str) -> bool:
      """Check if a database is blocked."""
      return db_name in settings.blocked_databases_list