cloudflare

The CF class: Cloudflare DNS records and tunnels through the official Python SDK

CF

CF holds one connection to the Cloudflare API and offers a method for each job. The token comes from the token argument, or from the CLOUDFLARE_API_TOKEN environment variable when you leave it out. timeout is the number of seconds to wait for each request, and any other keyword arguments go straight to the SDK’s Cloudflare client.

Every method returns plain dictionaries and lists, converted from the SDK’s objects. These results print and serialise without extra work.

The methods come in three groups:

• Zones and DNS records: zones, zone_id, dns_records, create_record, delete_record and upsert_record. A zone is a domain you manage on Cloudflare. The low-level methods take a zone ID; upsert_record takes the domain name and looks the ID up itself. Use it in most code.
• Tunnels: account_id, tunnels, tunnel_id, create_tunnel, delete_tunnel, tunnel_token and tunnel_cname. Tunnels belong to an account rather than a zone. tunnels, tunnel_id, create_tunnel, delete_tunnel and tunnel_token take an optional account_id and use your first account when you leave it out. tunnel_cname writes the DNS record that points a name at a tunnel. A CNAME cannot share a name with an A or AAAA record, so it deletes those two types at the name first and leaves every other type alone.
• Shortcuts: verify checks the token without changing anything, and setup_tunnel creates a tunnel, its DNS record and its token in one call.

upsert_record keeps one record of the given type per name and never touches records of other types, so a TXT or MX record beside an A record survives. Among the existing records of that type, one with the same content is kept, otherwise the first; it is updated in place when its content or proxied setting differs, and the other records of that type are deleted. If there is none, the record is created. A name is read as a subdomain of the domain unless it already ends in the domain; @ and the domain itself mean the apex, and a trailing dot is ignored.

---

source

CF

def CF(
    token:NoneType=None, timeout:int=10, **kw
):

A Cloudflare API client for DNS records and tunnels, built on the official cloudflare SDK.

Least-privilege token setup

Give the token only the permissions cfeasy needs. Create a custom token in the Cloudflare dashboard under My Profile -> API Tokens -> Create Token:

What it is for	Permission
DNS records	Zone -> DNS -> Edit (limit it to your zone)
Finding the account ID for tunnel calls	Account -> Account Settings -> Read
Creating, listing and deleting tunnels	Account -> Cloudflare Tunnel -> Edit

verify confirms the token works before you change anything. It checks that the token is active, that it can see an account, its zones (or the one zone named by domain) and that account’s tunnels. It returns ok, the token status, the account ID, the zone names, the tunnel count and an errors dict naming each check that failed. Account-owned tokens are checked against the account endpoint. The examples below ran against a real account and are not run by the docs build. The second setup_tunnel call serves the domain sankalpa.sh itself instead of a subdomain.

from types import SimpleNamespace as NS
from unittest.mock import MagicMock
from fastcore.test import test_eq, test_fail

class Pg(list):
    "SDK pages are truthy even when empty."
    def __bool__(self): return True

def fake_cf(status='active', zones=('example.com',), accts=('acc1',), user_tok=True):
    c = CF('x'); m = c._c = MagicMock()
    v = NS(id='t1', status=status)
    if user_tok: m.user.tokens.verify.return_value = v
    else: m.user.tokens.verify.side_effect = Exception('Invalid API Token')
    m.accounts.tokens.verify.return_value = v
    m.accounts.list.side_effect = lambda: Pg(NS(id=a) for a in accts)
    m.zones.list.side_effect = lambda name=None: Pg(NS(id=f'z-{z}', name=z, model_dump=lambda z=z: dict(name=z)) for z in zones if name in (None, z))
    m.zero_trust.tunnels.cloudflared.list.side_effect = lambda account_id: Pg()
    return c

r = fake_cf().verify()
test_eq(r, dict(ok=True, token='active', account='acc1', zones=['example.com'], tunnels=0, errors={}))
r = fake_cf(status='disabled').verify()
test_eq((r['ok'], r['token'], list(r['errors'])), (False, 'disabled', ['token']))
r = fake_cf(zones=()).verify()
test_eq((r['ok'], list(r['errors'])), (False, ['zones']))
r = fake_cf().verify(domain='other.com')
test_eq((r['ok'], r['zones'], r['errors']['zones']), (False, None, 'Zone not found: other.com'))
test_eq(fake_cf().verify(domain='example.com')['zones'], ['example.com'])
r = fake_cf(accts=()).verify()
test_eq((r['ok'], r['account'], r['tunnels'], 'account' in r['errors']), (False, None, None, True))
test_eq(fake_cf(user_tok=False).verify()['token'], 'active')

c = CF()
c.verify()

{'result': True}

tid, token = c.setup_tunnel('sankalpa.sh', 'app')   # creates DNS for app.sankalpa.sh
tid, token

('b59c4690-7c89-4ab7-9421-176afa6e2dfa',
 'eyJhIjoiOTg3ODZiMGRlMTM5OTE0YmVkNGMyYjdmNGYwYzg4YjkiLCJ0IjoiYjU5YzQ2OTAtN2M4OS00YWI3LTk0MjEtMTc2YWZhNmUyZGZhIiwicyI6ImVJN1RyWE1RK1BvU09nQUp2UG5uWHdzT0h0TFRlU1FMd1ZwU1lXSW1PVXhJWWxseWhnL21DeW1DRTJialRIcEM1QW90aGRkaXZzc3VCQ0RUSEdVZ3dnPT0ifQ==')

tid_apex, token_apex = c.setup_tunnel('sankalpa.sh')   # creates DNS for sankalpa.sh itself; Cloudflare flattens the CNAME at the root
tid_apex, token_apex

('6791fbaf-e848-422b-8fc7-35afc3e435d0',
 'eyJhIjoiOTg3ODZiMGRlMTM5OTE0YmVkNGMyYjdmNGYwYzg4YjkiLCJ0IjoiNjc5MWZiYWYtZTg0OC00MjJiLThmYzctMzVhZmMzZTQzNWQwIiwicyI6IjNNOVhaQm1IWkJPdDl3a28rRUxza3dmaWZ3ZUpqcmpSV0lxdDNRcExwTHFEZVJxaFRyeGRWZnR3M082TzJ1cFNOc0FYSXNRMGx0UjdhaE1SbVFxNDNnPT0ifQ==')

c.tunnel_id('app')                          # -> UUID
c.delete_tunnel(c.tunnel_id('app'))
c.delete_tunnel(c.tunnel_id('sankalpa.sh'))

Cache rules and purges

A ruleset is a list of Cloudflare rules. A phase identifies when Cloudflare applies a ruleset. The entrypoint ruleset is the ruleset that a zone configures for one phase. A zone keeps its cache rules and its custom firewall rules in entrypoint rulesets.

ruleset_upsert puts your rules into the entrypoint ruleset for phase. A ref is the stable identifier that ruleset_upsert uses to match a rule. Give each rule a stable ref. The function keeps every rule whose ref is absent from your input, with all its keys, such as logging or ratelimit. The update omits Cloudflare’s read-only fields version, last_updated and categories. The function compares the rules that have your refs with your rules on the keys description, expression, action, action_parameters and enabled, and on every other key that your rules set. It returns False and sends nothing when the compared keys match. Otherwise it writes the merged list and returns True. Running it on every deploy is safe. The token needs Zone -> Cache Rules -> Edit for the cache phase, and Zone -> Zone WAF -> Edit for http_request_firewall_custom.

purge empties the zone’s cache by URL files, by URL prefixes, or everything. Files and prefixes go in separate requests, files first, because the API takes one kind per request. tries limits the total number of attempts for one request. wait sets the delay in seconds before the first retry. Each later delay is twice the previous delay. The table shows how purge treats each failure.

Failure	Treatment
HTTP 408, HTTP 409, HTTP 429, Cloudflare code 1134, any HTTP 5xx, or a connection error such as a timeout	Retry, up to tries attempts
HTTP 403, or HTTP 400 whose message names the plan or a permission	Refusal: raise, or purge everything when fallback_all=True
Any other error, including a retryable failure that outlasts the retries	Raise and purge nothing more

By default, purge raises the refusal caused by a plan or permission restriction and purges nothing more. fallback_all defaults to False. Set fallback_all=True to request a full purge after the refusal. A full purge empties the whole zone cache, and the origin serves every request until the cache refills. The return value lists the purge requests that succeeded. After a fallback, the list holds the files purge when it ran first, then the purge of everything. A caller can tell from this list when the fallback ran.

---

source

CF.purge

def purge(
    zid, prefixes:NoneType=None, files:NoneType=None, everything:bool=False, fallback_all:bool=False, tries:int=4,
    wait:int=1
)->list:

Purge the zone cache by files, prefixes or everything, retrying rate limits, connection errors, HTTP 408, HTTP 409 and 5xx responses; purges everything when Cloudflare refuses a partial purge and fallback_all=True; returns the purges that succeeded.

---

source

CF.ruleset_upsert

def ruleset_upsert(
    zid, rules, phase:str='http_request_cache_settings'
)->bool:

Put rules in the zone entrypoint for phase, replacing the rules with the same ref and keeping the rest whole; False when nothing changed.

def fake_rules(rules=()):
    c, calls = CF('x'), []
    R = lambda d: NS(model_dump=lambda exclude_none=True: d)
    c._c = NS(rulesets=NS(phases=NS(get=lambda ph, zone_id: NS(rules=[R(r) for r in rules]),
                                    update=lambda ph, zone_id, rules: calls.append((ph, rules)))))
    return c, calls

mine = [dict(ref='a', expression='true', action='set_cache_settings', action_parameters=dict(cache=True), enabled=True)]
theirs = dict(id='o1', ref='theirs', expression='true', action='block', enabled=True, version='3', last_updated='2026-10-19T00:00:00Z',
              logging=dict(enabled=False), ratelimit=dict(characteristics=['ip.src'], period=60, requests_per_period=10))
c, calls = fake_rules([theirs, dict(mine[0], id='v1', expression='false')])
test_eq(c.ruleset_upsert('z', mine), True)
(ph, sent), = calls
test_eq((ph, [r['ref'] for r in sent]), ('http_request_cache_settings', ['theirs', 'a']))
test_eq(sent[0], {k: v for k, v in theirs.items() if k not in ('version', 'last_updated')})   # a rule it does not own keeps its whole body
assert 'version' not in sent[0]
c, calls = fake_rules([theirs, dict(mine[0], id='v1')])
test_eq((c.ruleset_upsert('z', mine, phase='http_request_firewall_custom'), calls), (False, []))
logged = [dict(mine[0], logging=dict(enabled=False))]
c, calls = fake_rules([dict(mine[0], id='v1')])
test_eq(c.ruleset_upsert('z', logged), True)   # a change to a key of its own beyond the usual ones is a change
bare = [dict(ref='b', expression='true', action='set_cache_settings', action_parameters=dict(cache=True), description=None)]
stored = dict(bare[0], id='v2', version='1', last_updated='2026-10-19T00:00:00Z', enabled=True, description='')   # the API fills both in
c, calls = fake_rules([stored])
test_eq((c.ruleset_upsert('z', bare), calls), (False, []))

import httpx
from unittest import mock
from cloudflare import APIConnectionError, APITimeoutError, BadRequestError, InternalServerError, PermissionDeniedError, RateLimitError
_rq = httpx.Request('POST', 'https://api.cloudflare.com/client/v4/zones/z/purge_cache')
def cf_err(cls, status, code, msg): return cls(msg, response=httpx.Response(status, request=_rq), body=dict(errors=[dict(code=code, message=msg)]))
refused = lambda: cf_err(PermissionDeniedError, 403, 10000, 'not on this plan')
def fake_purge(fails=None):
    'A client whose purge raises, for each kind of purge in `fails`, the errors listed there in turn, then succeeds; `calls` records each request.'
    c, calls, fails = CF('x'), [], {k: list(v) for k, v in (fails or {}).items()}
    def purge(zone_id, **kw):
        calls.append(kw)
        if (q := fails.get(first(kw))) and (e := q.pop(0)()): raise e
    c._c = NS(cache=NS(purge=purge))
    c._c.with_options = lambda **kw: c._c
    return c, calls

c, calls = fake_purge()
test_eq(c.purge('z', prefixes=['e.com/s/'], files=['https://e.com/']), [dict(files=['https://e.com/']), dict(prefixes=['e.com/s/'])])
test_eq(calls, [dict(files=['https://e.com/']), dict(prefixes=['e.com/s/'])])
c, calls = fake_purge(dict(prefixes=[refused]))
test_eq((c.purge('z', prefixes=['e.com/s/'], fallback_all=True), calls), ([dict(purge_everything=True)], [dict(prefixes=['e.com/s/']), dict(purge_everything=True)]))
c, calls = fake_purge(dict(prefixes=[refused]))
test_fail(lambda: c.purge('z', prefixes=['e.com/s/']), contains='not on this plan')
test_eq(calls, [dict(prefixes=['e.com/s/'])])   # the default sends no purge of everything
test_fail(lambda: fake_purge()[0].purge('z'), contains='purge needs')
test_eq(fake_purge()[0].purge('z', everything=True), [dict(purge_everything=True)])

# a refused prefixes purge after a files purge reports both; rate limits and timeouts retry after waits that double each time; any other error raises
c, calls = fake_purge(dict(prefixes=[refused]))
test_eq(c.purge('z', prefixes=['e.com/s/'], files=['https://e.com/'], fallback_all=True), [dict(files=['https://e.com/']), dict(purge_everything=True)])
waits, slow = [], lambda: cf_err(RateLimitError, 429, 1134, 'Unable to purge, rate limit reached')
with mock.patch('time.sleep', waits.append):
    c, calls = fake_purge(dict(prefixes=[slow, lambda: APITimeoutError(_rq)]))
    test_eq((c.purge('z', prefixes=['e.com/s/']), calls), ([dict(prefixes=['e.com/s/'])], [dict(prefixes=['e.com/s/'])] * 3))
    test_eq(waits, [1, 2])
    c, calls = fake_purge(dict(prefixes=[slow] * 4))
    test_fail(lambda: c.purge('z', prefixes=['e.com/s/'], tries=3), contains='rate limit')
    test_eq(len(calls), 3)   # no purge of everything after a rate limit
    c, calls = fake_purge(dict(prefixes=[lambda: cf_err(BadRequestError, 400, 1092, 'bad prefix'), lambda: 0]))
    test_fail(lambda: c.purge('z', prefixes=['https://e.com/s/']), contains='bad prefix')
    c, calls = fake_purge(dict(prefixes=[lambda: RuntimeError('socket closed'), lambda: 0]))
    test_fail(lambda: c.purge('z', prefixes=['e.com/s/']), contains='socket closed')
    test_eq(calls, [dict(prefixes=['e.com/s/'])])

# purge sends requests through `self._c`; a test client that attaches only `purge` can send purge requests
from unittest.mock import MagicMock
m, sent = MagicMock(), []
m._c = NS(cache=NS(purge=lambda zone_id, **kw: sent.append(kw)), with_options=lambda **kw: m._c)
test_eq((CF.purge(m, 'z', prefixes=['e.com/s/']), sent), ([dict(prefixes=['e.com/s/'])], [dict(prefixes=['e.com/s/'])]))

The SDK client retries a connection error, a rate limit and an HTTP 5xx response twice by default, and also HTTP 408 and 409. purge sends its requests through a client with max_retries=0, so tries is the total number of requests for one purge. purge retries all of these itself. The cell below counts the requests that reach a stand-in HTTP transport.

sent = []
def too_many(request):
    sent.append(request.url.path)
    body = dict(success=False, errors=[dict(code=1134, message='Unable to purge, rate limit reached')], messages=[], result=None)
    return httpx.Response(429, json=body, headers={'retry-after-ms': '1'})
c = CF('x', http_client=httpx.Client(transport=httpx.MockTransport(too_many)))
test_fail(lambda: c.purge('z', files=['https://e.com/'], tries=3, wait=0), contains='rate limit')
test_eq(len(sent), 3)
sent

purge also retries a connection error, an HTTP 5xx response, HTTP 408 and HTTP 409, which the SDK client retried before max_retries=0, and each retry counts against tries. The cell below sends a dropped connection and a 502, 408 and 409 response through the stand-in transport. Each purge makes tries requests and then raises.

def flaky(respond):
    sent = []
    def handler(request):
        sent.append(request.url.path)
        return respond(request)
    return sent, CF('x', http_client=httpx.Client(transport=httpx.MockTransport(handler)))
def drop(request): raise httpx.ConnectError('connection reset', request=request)
def status(code): return lambda request: httpx.Response(code, json=dict(success=False, errors=[], messages=[], result=None))
counts = {}
for name, respond, msg in [('connection error', drop, 'Connection error'), ('502', status(502), '502'), ('408', status(408), '408'), ('409', status(409), '409')]:
    sent, c = flaky(respond)
    test_fail(lambda: c.purge('z', files=['https://e.com/'], tries=3, wait=0), contains=msg)
    counts[name] = len(sent)
test_eq(counts, {'connection error': 3, '502': 3, '408': 3, '409': 3})
counts

An upsert touches only records of its own type. It keeps one record of that type under the name: the first record with the requested content, else the first record of that type. It updates the kept record in place when its content or proxied setting differs. It deletes the other records of that type under the name. It reads api.internal as a subdomain of the zone. The lookup and the new record then share one full name.

def rec(id, type, name, content, proxied=False): return NS(id=id, type=type, name=name, content=content, proxied=proxied, model_dump=lambda: dict(id=id, type=type, name=name, content=content, proxied=proxied))

c = fake_cf(); m = c._c
m.dns.records.list.side_effect = lambda zone_id, name: [rec('a', 'A', name, '1.1.1.1'), rec('t', 'TXT', name, 'v=spf1')]
m.dns.records.update.side_effect = lambda rid, **kw: rec(rid, kw['type'], kw['name'], kw['content'], kw['proxied'])
m.dns.records.create.side_effect = lambda **kw: rec('new', kw['type'], kw['name'], kw['content'], kw['proxied'])

c.upsert_record('example.com', 'www', '2.2.2.2')
test_eq(m.dns.records.delete.call_count, 0)
test_eq(m.dns.records.update.call_args.args, ('a',))

res = c.upsert_record('example.com', 'www', '1.1.1.1', proxied=True)
test_eq(res['proxied'], True)

c.upsert_record('example.com', 'api.internal', '3.3.3.3')
test_eq(m.dns.records.list.call_args.kwargs['name'], 'api.internal.example.com')
res

A CNAME cannot share a name with other address records, so tunnel_cname first deletes the A and AAAA records at the name. It leaves TXT, MX and every other type alone. An existing CNAME is updated in place, or reduced to one, by upsert_record. The check below uses a stand-in client that keeps its records in a list.

def fake_dns(*recs):
    c, store = fake_cf(), list(recs)
    def create(**kw): store.append(rec(f"n{len(store)}", kw['type'], kw['name'], kw['content'], kw['proxied'])); return store[-1]
    def delete(rid, zone_id): store[:] = [r for r in store if r.id != rid]
    c._c.dns.records.list.side_effect = lambda zone_id, name: [r for r in store if r.name == name]
    c._c.dns.records.create.side_effect = create
    c._c.dns.records.delete.side_effect = delete
    return c, store

c, store = fake_dns(rec('a', 'A', 'app.example.com', '1.1.1.1'), rec('aaaa', 'AAAA', 'app.example.com', '::1'), rec('t', 'TXT', 'app.example.com', 'v=spf1'), rec('o', 'A', 'other.example.com', '9.9.9.9'))
c.tunnel_cname('example.com', 'app', 'tid')
test_eq(sorted((r.type, r.name) for r in store), [('A', 'other.example.com'), ('CNAME', 'app.example.com'), ('TXT', 'app.example.com')])

c, store = fake_dns(rec('c1', 'CNAME', 'app.example.com', 'old.example.net'), rec('c2', 'CNAME', 'app.example.com', 'older.example.net'))
c.upsert_record('example.com', 'app', 'new.example.net', type='CNAME')
test_eq(len(store), 1)

c, store = fake_dns(rec('a1', 'A', 'example.com', '1.1.1.1'), rec('a2', 'A', 'example.com', '1.1.1.1'), rec('a3', 'A', 'example.com', '2.2.2.2'), rec('mx', 'MX', 'example.com', 'mail.example.com'))
c.upsert_record('example.com', 'example.com.', '1.1.1.1')
test_eq(sorted(r.id for r in store), ['a1', 'mx'])

c.upsert_record('example.com', '@', '3.3.3.3', type='TXT')
test_eq(sorted((r.type, r.name) for r in store), [('A', 'example.com'), ('MX', 'example.com'), ('TXT', 'example.com')])

c.tunnel_cname('example.com', 'example.com.', 'tid')
sorted((r.type, r.name) for r in store)

An error body whose errors is null still classifies, so the original failure is what reaches the caller.

from cloudflare import APIStatusError

err = APIStatusError('bad', response=httpx.Response(400, request=httpx.Request('POST', 'https://x.test')), body={'errors': None})
test_eq(_purge_kind(err), None)

Helpers

repo_root returns the top folder of the git repository you are working in, found by walking up from the current directory until it finds a .git entry. It returns None outside a repository.

cfeasy includes a SKILL.md file that tells coding agents such as Claude Code how to use the library. mv_skill_md copies it into a project’s .agents/skills/cfeasy/ and .claude/skills/cfeasy/ folders, where those agents look for skills. By default it only prints where it would copy the file; pass dry_run=False to copy it. The project root is dir if you give one, otherwise the repository root, otherwise the current directory.

---

source

mv_skill_md

def mv_skill_md(
    dry_run:bool=True, dir:NoneType=None
)->None:

Copy the bundled SKILL.md into the project’s agent skill folders, or with dry_run only print where it would go.

---

source

repo_root

def repo_root()->pathlib.Path:

Find the root of the current git repository, or None if not in a repo.