diff --git a/README.md b/README.md
index 8820ce9..8ba9e42 100644
--- a/README.md
+++ b/README.md
@@ -54,6 +54,9 @@ providers:
     #cdn: false
     # Manage Page Rules (URLFWD) records
     # pagerules: true
+    # Optional. Define Cloudflare plan type for the zones. Default: free,
+    # options: free, enterprise
+    #plan_type: free
     # Optional. Default: 4. Number of times to retry if a 429 response
     # is received.
     #retry_count: 4
diff --git a/octodns_cloudflare/__init__.py b/octodns_cloudflare/__init__.py
index 8dbda3e..e887708 100644
--- a/octodns_cloudflare/__init__.py
+++ b/octodns_cloudflare/__init__.py
@@ -27,8 +27,11 @@
 # TODO: remove __VERSION__ with the next major version release
 __version__ = __VERSION__ = '0.0.7'
 
+CLOUDFLARE_FREE_PLAN = 'free'
+
 
 class CloudflareError(ProviderException):
+
     def __init__(self, data):
         try:
             message = data['errors'][0]['message']
@@ -38,16 +41,19 @@ def __init__(self, data):
 
 
 class CloudflareAuthenticationError(CloudflareError):
+
     def __init__(self, data):
         CloudflareError.__init__(self, data)
 
 
 class CloudflareRateLimitError(CloudflareError):
+
     def __init__(self, data):
         CloudflareError.__init__(self, data)
 
 
 class Cloudflare5xxError(CloudflareError):
+
     def __init__(self, data):
         CloudflareError.__init__(self, data)
 
@@ -94,6 +100,7 @@ def __init__(
         account_id=None,
         cdn=False,
         pagerules=True,
+        plan_type=None,
         retry_count=4,
         retry_period=300,
         auth_error_retry_count=0,
@@ -105,11 +112,12 @@ def __init__(
     ):
         self.log = getLogger(f'CloudflareProvider[{id}]')
         self.log.debug(
-            '__init__: id=%s, email=%s, token=***, account_id=%s, cdn=%s',
+            '__init__: id=%s, email=%s, token=***, account_id=%s, cdn=%s, plan=%s',
             id,
             email,
             account_id,
             cdn,
+            plan_type,
         )
         super().__init__(id, *args, **kwargs)
 
@@ -128,6 +136,7 @@ def __init__(
         self.account_id = account_id
         self.cdn = cdn
         self.pagerules = pagerules
+        self.plan_type = plan_type
         self.retry_count = retry_count
         self.retry_period = retry_period
         self.auth_error_retry_count = auth_error_retry_count
@@ -231,7 +240,15 @@ def zones(self):
                 else:
                     page = None
 
-            self._zones = IdnaDict({f'{z["name"]}.': z['id'] for z in zones})
+            self._zones = IdnaDict(
+                {
+                    f'{z["name"]}.': {
+                        'id': z['id'],
+                        'plan': z.get('plan', {}).get('legacy_id', None),
+                    }
+                    for z in zones
+                }
+            )
 
         return self._zones
 
@@ -485,7 +502,7 @@ def _data_for_SSHFP(self, _type, records):
 
     def zone_records(self, zone):
         if zone.name not in self._zone_records:
-            zone_id = self.zones.get(zone.name, False)
+            zone_id = self.zones.get(zone.name, {}).get('id', False)
             if not zone_id:
                 return []
 
@@ -1033,7 +1050,7 @@ def _gen_key(self, data):
 
     def _apply_Create(self, change):
         new = change.new
-        zone_id = self.zones[new.zone.name]
+        zone_id = self.zones[new.zone.name]['id']
         if new._type == 'URLFWD':
             path = f'/zones/{zone_id}/pagerules'
         else:
@@ -1043,7 +1060,8 @@ def _apply_Create(self, change):
 
     def _apply_Update(self, change):
         zone = change.new.zone
-        zone_id = self.zones[zone.name]
+        zone_id = self.zones[zone.name]['id']
+
         hostname = zone.hostname_from_fqdn(change.new.fqdn[:-1])
         _type = change.new._type
 
@@ -1176,7 +1194,7 @@ def _apply_Delete(self, change):
         existing_name = existing.fqdn[:-1]
         # Make sure to map ALIAS to CNAME when looking for the target to delete
         existing_type = 'CNAME' if existing._type == 'ALIAS' else existing._type
-        zone_id = self.zones[existing.zone.name]
+        zone_id = self.zones[existing.zone.name]['id']
         for record in self.zone_records(existing.zone):
             if 'targets' in record and self.pagerules:
                 uri = record['targets'][0]['constraint']['value']
@@ -1209,23 +1227,71 @@ def _apply_Delete(self, change):
                     )
                     self._try_request('DELETE', path)
 
+    def _supported_plans(self, zone_name):
+        zone_id = self.zones[zone_name]['id']
+        path = f'/zones/{zone_id}/available_plans'
+        resp = self._try_request('GET', path)
+        result = resp['result']
+        if isinstance(result, list):
+            return [plan['legacy_id'] for plan in result]
+        msg = f'{self.id}: unable to determine supported plans, do you have an Enterprise account?'
+        raise SupportsException(msg)
+
+    def _update_plan(self, zone_name, plan):
+        current_plan = self.zones[zone_name].get('plan', False)
+        if current_plan and current_plan == plan:
+            return
+        if plan in self._supported_plans(zone_name):
+            zone_id = self.zones[zone_name]['id']
+            data = {'plan': {'legacy_id': plan}}
+            resp = self._try_request('PATCH', f'/zones/{zone_id}', data=data)
+            # Update the cached plan information
+            self.zones[zone_name]['plan'] = resp['result']['plan']['legacy_id']
+        else:
+            msg = f'{self.id}: {plan} is not supported for {zone_name}'
+            raise SupportsException(msg)
+
+    def _plan_meta(self, existing, desired, changes):
+        meta = {}
+        zone_name = desired.name
+        desired_plan = self.plan_type
+        if zone_name in self.zones:
+            current_plan = self.zones[zone_name].get('plan', False)
+            if desired_plan is None and current_plan is not False:
+                # If desired plan is removed and current plan is set, set to free plan
+                meta['cloudflare_plan'] = CLOUDFLARE_FREE_PLAN
+            elif current_plan != desired_plan:
+                meta['cloudflare_plan'] = desired_plan
+        return meta
+
     def _apply(self, plan):
         desired = plan.desired
         changes = plan.changes
+        zone_name = desired.name
+
         self.log.debug(
-            '_apply: zone=%s, len(changes)=%d', desired.name, len(changes)
+            '_apply: zone=%s, len(changes)=%d', zone_name, len(changes)
         )
 
-        name = desired.name
-        if name not in self.zones:
+        if zone_name not in self.zones:
             self.log.debug('_apply:   no matching zone, creating')
-            data = {'name': name[:-1], 'jump_start': False}
+            data = {'name': zone_name[:-1], 'jump_start': False}
             if self.account_id is not None:
                 data['account'] = {'id': self.account_id}
+            if self.plan_type is not None:
+                data['plan'] = {'legacy_id': self.plan_type}
             resp = self._try_request('POST', '/zones', data=data)
             zone_id = resp['result']['id']
-            self.zones[name] = zone_id
-            self._zone_records[name] = {}
+            self.zones[zone_name] = {'id': zone_id}
+            plan = resp['result'].get('plan', {}).get('legacy_id', False)
+            if plan is not False:
+                self.zones[zone_name]['plan'] = plan
+
+        # Handle plan changes if needed
+        # older versions of octodns don't have meta support.
+        meta = getattr(plan, 'meta', {})
+        if 'cloudflare_plan' in meta:
+            self._update_plan(zone_name, meta['cloudflare_plan'])
 
         # Force the operation order to be Delete() -> Create() -> Update()
         # This will help avoid problems in updating a CNAME record into an
@@ -1237,7 +1303,7 @@ def _apply(self, plan):
             getattr(self, f'_apply_{class_name}')(change)
 
         # clear the cache
-        self._zone_records.pop(name, None)
+        self._zone_records.pop(zone_name, None)
 
     def _extra_changes(self, existing, desired, changes):
         extra_changes = []
diff --git a/tests/test_octodns_provider_cloudflare.py b/tests/test_octodns_provider_cloudflare.py
index fb2321d..6a68a18 100644
--- a/tests/test_octodns_provider_cloudflare.py
+++ b/tests/test_octodns_provider_cloudflare.py
@@ -18,6 +18,7 @@
 from octodns.zone import Zone
 
 from octodns_cloudflare import (
+    CLOUDFLARE_FREE_PLAN,
     CloudflareAuthenticationError,
     CloudflareProvider,
     CloudflareRateLimitError,
@@ -1024,7 +1025,7 @@ def test_pagerules(self):
         # Set things up to preexist/mock as necessary
         zone = Zone('unit.tests.', [])
         # Stuff a fake zone id in place
-        provider._zones = {zone.name: '42'}
+        provider._zones = {zone.name: {'id': '42'}}
         provider._request = Mock()
         side_effect = [
             {
@@ -2911,6 +2912,309 @@ def test_process_desired_zone(self):
         self.assertTrue('subber.unit.tests.' in msg)
         self.assertTrue('coresponding NS record' in msg)
 
+    def test_meta(self):
+        provider = CloudflareProvider(
+            'test', 'email', 'token', 'account_id', plan_type='enterprise'
+        )
+        provider._try_request = Mock()
+
+        # Test 1: Creating new zone with plan_type
+        provider._try_request.side_effect = [
+            {
+                # GET /zones response (empty)
+                'result': [],
+                'result_info': {'count': 0, 'per_page': 50},
+            },
+            {
+                # POST /zones response
+                'result': {'id': '42', 'plan': {'legacy_id': 'enterprise'}},
+                'result_info': {'count': 1, 'per_page': 50},
+            },
+        ]
+
+        zone = Zone('unit.tests.', [])
+        plan = Plan(zone, zone, [], True)
+        provider._apply(plan)
+
+        provider._try_request.assert_has_calls(
+            [
+                call(
+                    'GET',
+                    '/zones',
+                    params={
+                        'page': 1,
+                        'per_page': 50,
+                        'account.id': 'account_id',
+                    },
+                ),
+                call(
+                    'POST',
+                    '/zones',
+                    data={
+                        'name': 'unit.tests',
+                        'jump_start': False,
+                        'account': {'id': 'account_id'},
+                        'plan': {'legacy_id': 'enterprise'},
+                    },
+                ),
+            ]
+        )
+
+        self.assertEqual(
+            provider.zones['unit.tests.'], {'id': '42', 'plan': 'enterprise'}
+        )
+
+        # Reset for next test
+        provider._try_request.reset_mock()
+
+        # Test 2: Plan update when current plan differs
+        provider._zones = {'unit.tests.': {'id': '42', 'plan': 'pro'}}
+
+        existing = Zone('unit.tests.', [])
+        desired = Zone('unit.tests.', [])
+
+        provider._try_request.side_effect = [
+            {'result': [{'legacy_id': 'pro'}, {'legacy_id': 'enterprise'}]},
+            {
+                # PATCH /zones/42 (plan update)
+                'result': {'plan': {'legacy_id': 'enterprise'}}
+            },
+        ]
+
+        meta = provider._plan_meta(existing, desired, [])
+        self.assertEqual({'cloudflare_plan': 'enterprise'}, meta)
+        plan = Plan(existing, desired, [], True, meta=meta)
+        provider.apply(plan)
+
+        provider._try_request.assert_has_calls(
+            [
+                call('GET', '/zones/42/available_plans'),
+                call(
+                    'PATCH',
+                    '/zones/42',
+                    data={'plan': {'legacy_id': 'enterprise'}},
+                ),
+            ]
+        )
+
+        # Test 3: Plan removal sets to free plan
+        provider = CloudflareProvider(
+            'test', 'email', 'token', 'account_id', plan_type=None
+        )
+        provider._zones = {'unit.tests.': {'id': '42', 'plan': 'pro'}}
+        provider._try_request = Mock()
+        provider._try_request.side_effect = [
+            {
+                'result': [
+                    {'legacy_id': 'pro'},
+                    {'legacy_id': CLOUDFLARE_FREE_PLAN},
+                ]
+            },
+            {'result': {'plan': {'legacy_id': CLOUDFLARE_FREE_PLAN}}},
+        ]
+
+        meta = provider._plan_meta(existing, desired, [])
+        self.assertEqual({'cloudflare_plan': CLOUDFLARE_FREE_PLAN}, meta)
+        plan = Plan(existing, desired, [], True, meta=meta)
+        provider.apply(plan)
+
+        provider._try_request.assert_has_calls(
+            [
+                call('GET', '/zones/42/available_plans'),
+                call(
+                    'PATCH',
+                    '/zones/42',
+                    data={'plan': {'legacy_id': CLOUDFLARE_FREE_PLAN}},
+                ),
+            ]
+        )
+
+        # Test 4: No plan update when current plan matches desired plan
+        provider = CloudflareProvider(
+            'test', 'email', 'token', 'account_id', plan_type='enterprise'
+        )
+        provider._zones = {'unit.tests.': {'id': '42', 'plan': 'enterprise'}}
+        provider._try_request = Mock()
+
+        meta = provider._plan_meta(existing, desired, [])
+        self.assertEqual({}, meta)  # No meta changes when plans match
+
+        # Test 5: Unsupported plan raises SupportsException
+        provider = CloudflareProvider(
+            'test', 'email', 'token', 'account_id', plan_type='unsupported_plan'
+        )
+        provider._zones = {'unit.tests.': {'id': '42', 'plan': 'pro'}}
+        provider._try_request = Mock()
+        provider._try_request.side_effect = [
+            {'result': [{'legacy_id': 'pro'}, {'legacy_id': 'enterprise'}]}
+        ]
+
+        meta = provider._plan_meta(existing, desired, [])
+        self.assertEqual({'cloudflare_plan': 'unsupported_plan'}, meta)
+        plan = Plan(existing, desired, [], True, meta=meta)
+
+        with self.assertRaises(SupportsException) as ctx:
+            provider.apply(plan)
+
+        self.assertEqual(
+            'test: unsupported_plan is not supported for unit.tests.',
+            str(ctx.exception),
+        )
+
+        provider._try_request.assert_called_once_with(
+            'GET', '/zones/42/available_plans'
+        )
+
+        # Test 6: No API calls when current plan matches desired plan
+        provider = CloudflareProvider(
+            'test', 'email', 'token', 'account_id', plan_type='enterprise'
+        )
+        provider._zones = {'unit.tests.': {'id': '42', 'plan': 'enterprise'}}
+        provider._try_request = Mock()
+
+        meta = provider._plan_meta(existing, desired, [])
+        self.assertEqual({}, meta)  # No meta changes when plans match
+        plan = Plan(existing, desired, [], True, meta=meta)
+        provider.apply(plan)
+
+        provider._try_request.assert_not_called()  # No API calls should be made
+
+        # Test 7: Enterprise check error handling
+        provider = CloudflareProvider(
+            'test', 'email', 'token', 'account_id', plan_type='enterprise'
+        )
+        provider._zones = {'unit.tests.': {'id': '42', 'plan': 'pro'}}
+        provider._try_request = Mock()
+        provider._try_request.side_effect = [
+            {'result': None}
+        ]  # Invalid response
+
+        meta = provider._plan_meta(existing, desired, [])
+        self.assertEqual({'cloudflare_plan': 'enterprise'}, meta)
+        plan = Plan(existing, desired, [], True, meta=meta)
+
+        with self.assertRaises(SupportsException) as ctx:
+            provider.apply(plan)
+
+        self.assertEqual(
+            'test: unable to determine supported plans, do you have an Enterprise account?',
+            str(ctx.exception),
+        )
+
+        provider._try_request.assert_called_once_with(
+            'GET', '/zones/42/available_plans'
+        )
+
+        # Test 8: Early return when current plan matches desired plan
+        provider = CloudflareProvider(
+            'test', 'email', 'token', 'account_id', plan_type='enterprise'
+        )
+        provider._zones = {'unit.tests.': {'id': '42', 'plan': 'enterprise'}}
+        provider._try_request = Mock()
+
+        meta = provider._plan_meta(existing, desired, [])
+        self.assertEqual({}, meta)  # No meta changes when plans match
+        plan = Plan(existing, desired, [], True, meta=meta)
+        provider.apply(plan)
+
+        provider._try_request.assert_not_called()  # No API calls should be made
+
+        # Test 9: Unsupported plan raises SupportsException
+        provider = CloudflareProvider(
+            'test', 'email', 'token', 'account_id', plan_type='unsupported_plan'
+        )
+        provider._zones = {'unit.tests.': {'id': '42', 'plan': 'pro'}}
+        provider._try_request = Mock()
+        provider._try_request.side_effect = [
+            {'result': [{'legacy_id': 'pro'}, {'legacy_id': 'enterprise'}]}
+        ]
+
+        meta = provider._plan_meta(existing, desired, [])
+        self.assertEqual({'cloudflare_plan': 'unsupported_plan'}, meta)
+        plan = Plan(existing, desired, [], True, meta=meta)
+
+        with self.assertRaises(SupportsException) as ctx:
+            provider.apply(plan)
+
+        self.assertEqual(
+            'test: unsupported_plan is not supported for unit.tests.',
+            str(ctx.exception),
+        )
+
+        provider._try_request.assert_called_once_with(
+            'GET', '/zones/42/available_plans'
+        )
+
+        # Test 10: Available plans API call failure
+        provider = CloudflareProvider(
+            'test', 'email', 'token', 'account_id', plan_type='enterprise'
+        )
+        provider._zones = {'unit.tests.': {'id': '42', 'plan': 'pro'}}
+        provider._try_request = Mock()
+        provider._try_request.side_effect = [
+            {'result': None}
+        ]  # Invalid response
+
+        meta = provider._plan_meta(existing, desired, [])
+        self.assertEqual({'cloudflare_plan': 'enterprise'}, meta)
+        plan = Plan(existing, desired, [], True, meta=meta)
+
+        with self.assertRaises(SupportsException) as ctx:
+            provider.apply(plan)
+
+        self.assertEqual(
+            'test: unable to determine supported plans, do you have an Enterprise account?',
+            str(ctx.exception),
+        )
+
+        provider._try_request.assert_called_once_with(
+            'GET', '/zones/42/available_plans'
+        )
+
+        # Test 11: Early return in _update_plan when plans match
+        provider = CloudflareProvider(
+            'test', 'email', 'token', 'account_id', plan_type='enterprise'
+        )
+        provider._zones = {'unit.tests.': {'id': '42', 'plan': 'enterprise'}}
+        provider._supported_plans = Mock()  # This shouldn't be called
+        provider._try_request = Mock()  # This shouldn't be called
+
+        # Call _update_plan directly with matching plans
+        provider._update_plan('unit.tests.', 'enterprise')
+
+        # Verify no API calls were made since plans match
+        provider._supported_plans.assert_not_called()
+        provider._try_request.assert_not_called()
+
+        # Test 12: Plan meta when zone doesn't exist
+        provider = CloudflareProvider(
+            'test', 'email', 'token', 'account_id', plan_type='enterprise'
+        )
+        provider._zones = {}  # No zones at all
+        provider._try_request = Mock()
+
+        meta = provider._plan_meta(existing, desired, [])
+        self.assertEqual({}, meta)
+        plan = Plan(existing, desired, [], True, meta=meta)
+
+        provider._try_request.side_effect = [
+            {
+                'result': {'id': '42', 'plan': {'legacy_id': 'enterprise'}},
+                'result_info': {'count': 1, 'per_page': 50}
+            }
+        ]
+
+        provider.apply(plan)
+
+        provider._try_request.assert_has_calls([
+            call('POST', '/zones', data={
+                'name': 'unit.tests',
+                'jump_start': False,
+                'account': {'id': 'account_id'},
+                'plan': {'legacy_id': 'enterprise'}
+            })
+        ])
+
 
 # temporarily override Plan to add meta attribute, until octodns dependency is updated
 # with https://github.com/octodns/octodns/pull/1236