CockroachDB MCP Server

Instructions

Args:
    schema: Filter by schema name (default: all user schemas).
    include_views: Include views in results.
    include_system: Include system tables.

Returns:
    List of tables with schema, name, and type.

Implementation Reference

• src/cockroachdb_mcp/tools/tables.py:118-193 (handler)

  Core implementation of list_tables tool: executes SQL query on information_schema.tables, applies filters for types (tables/views), schemas, system objects, and allowed schemas, processes and returns structured list of tables.

  async def list_tables(
      schema: str | None = None,
      include_views: bool = True,
      include_system: bool = False,
  ) -> dict[str, Any]:
      """List all tables in the database.
  
      Args:
          schema: Filter by schema name.
          include_views: Include views in results.
          include_system: Include system tables.
  
      Returns:
          List of tables.
      """
      conn = await connection_manager.ensure_connected()
  
      try:
          # Build type filter
          table_types = ["'BASE TABLE'"]
          if include_views:
              table_types.append("'VIEW'")
  
          type_filter = f"table_type IN ({','.join(table_types)})"
  
          # Build schema filter
          schema_filter = ""
          if schema:
              schema_filter = f"AND table_schema = '{schema}'"
          elif not include_system:
              schema_filter = """
                  AND table_schema NOT IN (
                      'crdb_internal', 'information_schema', 'pg_catalog', 'pg_extension'
                  )
              """
  
          query = f"""
              SELECT
                  table_schema,
                  table_name,
                  table_type
              FROM information_schema.tables
              WHERE {type_filter}
              {schema_filter}
              ORDER BY table_schema, table_name
          """
  
          async with conn.cursor() as cur:
              await cur.execute(query)
              rows = await cur.fetchall()
  
          tables = []
          for row in rows:
              schema_name = row.get("table_schema", "")
  
              # Check if schema is allowed
              if not _is_allowed_schema(schema_name):
                  continue
  
              tables.append(
                  {
                      "schema": schema_name,
                      "name": row.get("table_name", ""),
                      "type": "VIEW" if row.get("table_type") == "VIEW" else "TABLE",
                      "full_name": f"{schema_name}.{row.get('table_name', '')}",
                  }
              )
  
          return {
              "tables": tables,
              "count": len(tables),
              "database": connection_manager.current_database,
              "schema_filter": schema,
          }
      except Exception as e:
          return {"status": "error", "error": str(e)}

• src/cockroachdb_mcp/server.py:151-170 (registration)

  MCP tool registration for 'list_tables' using @mcp.tool() decorator. Defines input schema via type hints and docstring, wraps the core handler from tables module with error handling.

  @mcp.tool()
  async def list_tables(
      schema: str | None = None,
      include_views: bool = True,
      include_system: bool = False,
  ) -> dict[str, Any]:
      """List all tables in the database.
  
      Args:
          schema: Filter by schema name (default: all user schemas).
          include_views: Include views in results.
          include_system: Include system tables.
  
      Returns:
          List of tables with schema, name, and type.
      """
      try:
          return await tables.list_tables(schema, include_views, include_system)
      except Exception as e:
          return {"status": "error", "error": str(e)}