consul-http

Lightweight Consul KV client built on httpx.

中文说明见下方。

Why this package?

Unlike full-featured clients such as py-consul / python-consul, this library focuses on:

• In scope: KV (get / put / delete / prefix), Session, distributed lock, blocking watch_kv
• Out of scope (for now): Agent / Catalog / Health / Txn / Connect — use a full Consul SDK if you need those
• First-class sync + async with the same API shape
• Async-friendly async with client.lock(...) for leader election
• Optional injection of an existing httpx.Client / AsyncClient (shared pools, proxies, OTel)
• Small dependency surface (httpx only)
• Explicit typing (py.typed)

Compatibility

Targeted at Consul 1.10+ HTTP API (KV + Session).

URL / port rules (intentional):

Install

pip install consul-http
# or
uv add consul-http

Editable (from this monorepo):

uv add --editable ./packages/consul_http

Requires Python 3.10+.

Tests

# unit (default; skips live Consul)
uv run pytest

# live agent — needs write ACL; loads CONSUL_HTTP_* from repo .env when present
$env:CONSUL_INTEGRATION = "1"   # PowerShell
uv run pytest -m integration

Quick start

from consul_http import ConsulClient

# Uses CONSUL_HTTP_ADDR / CONSUL_HTTP_TOKEN when set
with ConsulClient() as c:
    c.put_kv("app/demo/key", "hello")
    res = c.get_kv("app/demo/key")
    if res is None:
        print("missing")
    else:
        print(res.raw_value, res.index)
    c.delete_kv("app/demo/key")

Async + blocking query

import asyncio
from consul_http import AsyncConsulClient

async def watch(key: str) -> None:
    async with AsyncConsulClient() as c:
        res = await c.get_kv(key)
        assert res is not None
        index = res.index
        while True:
            res = await c.get_kv(key, index=index, wait="30s")
            if res is None:
                continue
            if res.index != index:
                print("changed:", res.raw_value)
                index = res.index

# asyncio.run(watch("app/demo/key"))

Prefer the built-in watcher (handles index regression after leader change / snapshot restore — waiting on a stale index would hang forever otherwise):

async def watch(key: str) -> None:
    async with AsyncConsulClient() as c:
        async for res in c.watch_kv(key, wait="30s"):
            if res is None:
                print("missing")
                continue
            print("changed:", res.raw_value)

Session + distributed lock

from consul_http import AsyncConsulClient

async def on_lost() -> None:
    print("lock lost — stop critical work")

async def run_leader() -> None:
    async with AsyncConsulClient() as c:
        lock = c.lock(
            "service/demo/leader",
            ttl="15s",
            value="instance-1",
            on_lost=on_lost,  # sync or async callable
        )
        async with lock as acquired:
            if not acquired:
                return
            while lock.is_held:
                # do leader work; exit early if renew failed
                ...

# Low-level: create_session / put_kv(..., acquire=...) / renew_session / destroy_session
# Read lock holder: c.get_kv("service/demo/leader", raw=False) → res.session, res.lock_index

Sync mirror: with client.lock(...) as acquired: (on_lost must be sync there).

Exit skips KV release when the lock was already lost (lock.lost is True).

Injected httpx client / per-request token

import httpx
from consul_http import ConsulClient

# Share pools / proxies / custom transport; ConsulClient will NOT close `http`.
# Default headers (incl. X-Consul-Token) and connect settings stay on `http` —
# constructing ConsulClient(client=...) does not merge token into that client.
# Blocking / watch reads still apply ConsulClient.timeout via per-request timeout.
with httpx.Client(
    base_url="http://127.0.0.1:8500/v1",
    headers={"X-Consul-Token": "default-tok"},
    timeout=30.0,
) as http:
    with ConsulClient(client=http, timeout=10.0) as c:
        c.get_kv("app/demo/key", token="one-shot-acl-token")

Binary KV

with ConsulClient() as c:
    c.put_kv("app/demo/blob", b"\x00\xffprotobuf-or-msgpack")
    res = c.get_kv("app/demo/blob", raw=False)
    assert res is not None
    data = res.raw_bytes  # exact bytes; raw_value may use U+FFFD on bad UTF-8

Retry / backoff

By default clients retry transient failures up to 3 attempts with exponential backoff + jitter (RetryConfig).

CAS / lock / session-create caveat: For put_kv / delete_kv with cas= / acquire= / release=, and for create_session, status-code retries are disabled so a lost 502/504 after a successful apply is not re-issued (which could look like a false conflict / failed acquire, or orphan extra sessions). Transport-error retries still run; if a request may have been applied but the response was lost, a later retry can still return false / ConsulCASConflictError, or create an extra session (TTL eventually reaps orphans). When that happens and you suspect a retry race, re-get_kv (prefer raw=False) and compare value / session before treating it as a real concurrent write.

from consul_http import ConsulClient, RetryConfig

# Custom policy
with ConsulClient(retry=RetryConfig(max_attempts=5, backoff_factor=0.2)) as c:
    c.get_kv("app/demo/key")

# Disable retries
with ConsulClient(retry=RetryConfig(max_attempts=1)) as c:
    c.get_kv("app/demo/key")
# or: ConsulClient(retry=None)

Writes do not retry transport errors by default because the server may have applied a write before the response was lost. Such cases raise ConsulRequestOutcomeUnknown; reconcile the key before deciding whether to retry. RetryConfig(respect_retry_after=True) honors a server Retry-After header for retryable status codes.

Production watch lifecycle

from threading import Event

stop = Event()
with ConsulClient() as c:
    for update in c.watch_kv(
        "app/config",
        stop_event=stop,
        max_consecutive_failures=5,
    ):
        if update is not None:
            apply_config(update.value)

Use raise_if_lost() during long lock-protected work. Lock loss cannot fence an already-running process from writing an external system; use a fencing token at the downstream storage layer when that guarantee is required.

For observability, pass an implementation of EventSink to the client. Hook exceptions are isolated from request execution and events never contain ACL tokens or KV values.

Environment variables

Errors

• ConsulConnectionError — transport / timeout failures
• ConsulPermissionError — HTTP 401/403 (invalid token / ACL deny); subclass of ConsulAPIError
• ConsulAPIError — other non-success HTTP (body truncated in message)
• ConsulCASConflictError — put_kv/delete_kv with cas= returned false when raise_on_cas_conflict=True

中文

轻量 Consul KV 客户端（httpx Sync/Async），含 Session / 分布式锁。默认指数退避重试；cas/acquire/release/create_session 不重试 HTTP 状态码。目标 API：Consul 1.10+。

from consul_http import ConsulClient

with ConsulClient.from_base_url("http://127.0.0.1:8500") as c:
    print(c.get_kv("app/demo/key"))