lkrequest

A Python HTTP client with TLS/HTTP2/TCP fingerprint control. Powered by Rust for high performance.

English | 简体中文

Features

• Browser fingerprint emulation — TLS/HTTP2/TCP fingerprint presets for Chrome, Firefox, and Safari
• Custom fingerprints — Fully programmable TlsProfile / H2Profile / TcpFingerprint, with JSON serialization
• Fingerprint randomization — Client(randomize=Randomize.extension_order()) permutes TLS extension order per connection (drifts JA3, keeps JA4 stable, still a real browser); client.randomize_fingerprint() generates a unique variant to reduce correlation risk
• Client pool — ClientPool automatically rotates across multiple fingerprints
• Async & sync APIs — Both asyncio async and traditional synchronous calls
• WebSocket — wss:// connections with consistent fingerprinting
• Streaming responses — send_streaming() receives large files chunk by chunk
• Cookie management — Automatic cookie jar, with manual management, attribute setting, and overrides
• Proxy support — HTTP CONNECT / SOCKS5 proxies, authentication, and ordered multi-hop chains; QUIC/H3 works over all-SOCKS5 chains
• Proxy pool & session pool — Built-in proxy rotation, bad-proxy marking, and session management
• Multipart — File upload support
• Retry strategies — Exponential backoff / fixed interval / custom callable retries
• Middleware — Request/response interception and modification (onion model)
• Event hooks — Lightweight request/response callbacks
• Connection prewarming — preconnect() / prefetch() to batch-establish connections
• Prometheus metrics — Built-in request counters, latency histograms, exposition-format export
• Request diagnostics — response.diagnostics exposes per-request phase timings (DNS/TCP/TLS/TTFB/total) plus remote_addr/protocol/cipher_suite
• Runtime metrics snapshot — metrics_snapshot() pulls process-level wire bytes / connection / request counters (for host-side collection; real values require the telemetry feature)
• Zero-copy Response — bytes::Bytes + PyBuffer protocol + text/json caching
• Automatic decompression — Brotli / gzip / deflate / zstd, with controllable AcceptEncoding
• Certificate management — Custom CA / disable verification / system certificates
• ECH — Encrypted Client Hello support, including TLS/QUIC HelloRetryRequest handling
• Request priority — RequestPriority (RFC 9218 urgency/incremental), session.get(url, priority=...)
• Protocol policy — ProtocolPolicy / HttpIntent control H2/H3 selection, acquisition, and fallback (client / session / request level)
• Session resumption control — SessionResumptionConfig controls TLS1.3 PSK / TLS1.2 ticket resumption
• Request-level protocol override — preferred_http_version / idempotency (0-RTT replay-safety declaration)
• QUIC / HTTP3 — Optional feature (maturin develop --features quic-h3): session(http3_only=True / http3_with_fallback=True / broken_quic_policy=BrokenQuicPolicy.Resilient); Client(quic_fingerprint=..., quic_profile="chrome_150", disable_http3=True); dedicated Chrome 146/150 QuicProfile presets (available only when the feature is enabled)
• Synthetic fingerprints (advanced) — Optional feature (maturin develop --features synthetic-fp): Client(randomize=Randomize.recombine()) synthesizes a cross-layer (TLS+H2+H3) unique identity per session; Randomize.full() additionally draws out-of-corpus values for H2/QUIC; the Layers mask (e.g. Randomize.recombine_layers(Layers.TLS | Layers.H2)) restricts which layers are synthesized. Synthetic fingerprints match no real browser and are only for blocklist (negative-model) targets — against an allowlist they fail instantly

Installation

Using pip:

pip install lkrequest

In a project managed by uv:

uv add lkrequest

Development environment

Using venv and pip:

git clone https://github.com/EZ-XLabs/lkrequest-py
cd lkrequest-py
python3 -m venv .venv                 # Linux/macOS
# py -m venv .venv                    # Windows
source .venv/bin/activate             # Linux/macOS
# .venv\Scripts\Activate.ps1          # Windows PowerShell
python -m pip install --upgrade pip
python -m pip install maturin
python -m pip install -e ".[test,lint]"

Using uv (creates and manages .venv automatically):

git clone https://github.com/EZ-XLabs/lkrequest-py
cd lkrequest-py
uv sync --extra test --extra lint

Quick Start

Async API

import asyncio
import lkrequest

async def main():
    client = lkrequest.Client.chrome_144()
    session = client.session()

    # GET request
    resp = await session.get("https://httpbin.org/get", params={"key": "value"})
    print(resp.status_code)  # 200
    print(resp.json())

    # POST JSON
    resp = await session.post("https://httpbin.org/post", json={"hello": "world"})
    print(resp.json())

    # Concurrent requests
    urls = [f"https://httpbin.org/get?id={i}" for i in range(5)]
    responses = await asyncio.gather(*[session.get(url) for url in urls])

asyncio.run(main())

Sync API

from lkrequest.blocking import Client

client = Client.chrome_131()
session = client.session()

resp = session.get("https://httpbin.org/get")
print(resp.status_code)
print(resp.text())
print(resp.json())

All HTTP methods

session.get(url)
session.post(url, json={"key": "value"})
session.put(url, json={"key": "value"})
session.patch(url, json={"field": "new_value"})
session.delete(url)
session.head(url)
session.options(url)

Authentication

# Bearer Token
resp = session.get("https://httpbin.org/bearer", bearer_auth="my-token")

# Basic Auth
resp = session.get(
    "https://httpbin.org/basic-auth/user/pass",
    basic_auth=("user", "pass"),
)

Custom client

client = lkrequest.Client(
    tls_profile="chrome_144",
    h2_profile="chrome_144",
    tcp_fingerprint="chrome_win",
    default_headers={"Accept-Language": "zh-CN"},
    header_order=["Host", "User-Agent", "Accept"],
    total_timeout=30.0,
    tcp_connect_timeout=10.0,
    max_connections_per_session=16,
    h2_fallback_h1=True,
    proxy_fallback_direct=True,
)

Retry strategies

# Exponential backoff retry
session = client.session(
    retry=lkrequest.ExponentialBackoff(max_retries=3, base_delay=0.5, max_delay=30.0, jitter=True)
)

# Fixed interval retry
session = client.session(
    retry=lkrequest.FixedInterval(max_retries=5, interval=1.0)
)

# Custom retry strategy (callable)
def my_retry(attempt: int, error: str | None, status: int | None) -> float | None:
    if status == 429:
        return min(2 ** attempt, 30)
    if attempt < 3:
        return 1.0
    return None  # give up

session = client.session(retry=my_retry)

Middleware

def log_request(req_dict):
    print(f"{req_dict['method']} {req_dict['url']}")
    return req_dict

def log_response(resp_dict):
    print(f"  → {resp_dict['status']}")
    return resp_dict

# Client-level middleware
client = lkrequest.Client(
    middleware=[lkrequest.Middleware("logger", on_request=log_request, on_response=log_response)]
)

# Session-level middleware
session = client.session(
    middleware=[lkrequest.Middleware("auth", on_request=inject_auth)]
)

Event hooks

A lighter-weight callback mechanism than middleware, for observing requests/responses without modifying them.

def on_req(method, url, headers):
    print(f"→ {method} {url}")

def on_resp(status_code, url, elapsed):
    print(f"← {status_code} {url} ({elapsed:.3f}s)")

# Bind when creating the session
session = client.session(on_request=on_req, on_response=on_resp)

# Or add dynamically
session.on_request(on_req)
session.on_response(on_resp)

Cookie management

# Basic operations
session.set_cookie("https://example.com", "token", "abc123")
val = session.get_cookie("https://example.com", "token")
cookies = session.get_cookies("https://example.com")
header = session.cookie_header("https://example.com")

# With attributes
session.set_cookie_with_attrs(
    "https://example.com", "secure_token", "xyz",
    path="/api", domain="example.com", secure=True, http_only=True,
)

# Remove and clear
session.remove_cookie("https://example.com", "token")
session.clear_cookies()

# Request-level cookie override
resp = session.get(url, cookie_override={"token": "override_value"})

Multipart file upload

mp = lkrequest.Multipart()
mp.text("title", "File upload")
mp.file("document", "report.pdf", "application/pdf", pdf_bytes)
resp = session.post("https://example.com/upload", multipart=mp)

Streaming responses

# Async
stream = await session.send_streaming("GET", "https://example.com/large-file")
async for chunk in stream:
    process(chunk)

# Or read all at once
body = await stream.bytes()
text = await stream.text()

Connection prewarming

Establish TLS connections ahead of time to reduce first-request latency.

# Batch prewarming
results = await session.prefetch([
    "https://api.example.com",
    "https://cdn.example.com",
])
for r in results:
    print(f"{r['url']}: {'ok' if r['success'] else r['error']} ({r['duration_ms']:.0f}ms)")

# Single-connection prewarming
await session.preconnect("https://api.example.com")

# Sync version
session.preconnect("https://api.example.com")
results = session.prefetch(["https://api.example.com"])

WebSocket

# Async
ws = await session.ws_connect("wss://echo.websocket.events")
await ws.send_text("hello")
msg = await ws.recv()
async for msg in ws:
    print(msg)
    break
await ws.close()

# Sync
ws = session.ws_connect("wss://echo.websocket.events")
ws.send_text("hello")
msg = ws.recv()
ws.close()

Proxy

# Single proxy
session = client.session(proxy="socks5://user:pass@host:1080")

# Multi-hop chain: client -> hop1 -> hop2 -> target
chain = lkrequest.ProxyConfig.parse_chain([
    "socks5://user:pass@hop1:1080",
    "socks5h://user:pass@hop2:1080",
])
session = client.session(proxy=chain, http3_with_fallback=True)

# Request-level proxy override
resp = await session.get("https://example.com", proxy="http://other-proxy:8080")

# Proxy pool
pool = lkrequest.ProxyPool(
    ["socks5://proxy1:1080", "socks5://proxy2:1080", "http://proxy3:8080"],
    rotation="round_robin",  # or "random"
    bad_proxy_config=lkrequest.BadProxyConfig(
        failure_threshold=5, window=120.0, cooldown_duration=300.0, max_cooldowns=5,
    ),
    health_check=lkrequest.HealthCheckConfig(
        interval=30.0, timeout=3.0, target_host="www.google.com", target_port=443,
    ),
)
proxy = await pool.acquire()

TCP-based HTTP supports mixed HTTP CONNECT and SOCKS5 hops. QUIC/HTTP3 requires every hop in the chain to be SOCKS5; a chain containing an HTTP hop falls back to H2 when http3_with_fallback=True.

Session pool

# Async
pool = lkrequest.SessionPool(client=client, proxies=[...], max_sessions=10)
guard = await pool.acquire()
async with guard as session:
    resp = await session.get("https://example.com")

# Sync
from lkrequest.blocking import Client, SessionPool
pool = SessionPool(client=client, proxies=[...])
with pool.acquire() as session:
    resp = session.get("https://example.com")

Client pool

Automatically rotate across multiple fingerprints to reduce fingerprint correlation.

pool = lkrequest.ClientPool(
    [lkrequest.Client.chrome_144(), lkrequest.Client.firefox_147(), lkrequest.Client.safari_26()],
    rotation="round_robin",  # or "random"
)
client = pool.acquire()
pool.add(lkrequest.Client.chrome_131())  # add dynamically
print(len(pool))  # 4

Custom fingerprints

Fully programmable TLS, HTTP/2, and TCP fingerprint configuration.

# Custom TLS Profile
tls = lkrequest.TlsProfile(
    "my_browser",
    cipher_suites=[0x1301, 0x1302, 0x1303, 0xc02b, 0xc02f],
    extensions=[
        lkrequest.ExtensionSpec(lkrequest.ExtType.SNI),
        lkrequest.ExtensionSpec(lkrequest.ExtType.ALPN),
        lkrequest.ExtensionSpec(lkrequest.ExtType.SUPPORTED_GROUPS),
        lkrequest.ExtensionSpec(lkrequest.ExtType.KEY_SHARE),
        lkrequest.ExtensionSpec(lkrequest.ExtType.SUPPORTED_VERSIONS),
        lkrequest.ExtensionSpec(lkrequest.ExtType.SIGNATURE_ALGORITHMS),
    ],
    alpn_protocols=["h2", "http/1.1"],
    grease=lkrequest.GreaseConfig(extensions=True, key_share=True),
    padding=lkrequest.PaddingStrategy.block_align(128, 512),
)

# Custom H2 Profile
h2 = lkrequest.H2Profile(
    [
        lkrequest.H2Setting("header_table_size", 65536),
        lkrequest.H2Setting("initial_window_size", 6291456),
        lkrequest.H2Setting("max_header_list_size", 262144),
    ],
    15663105,  # window_update
    ["method", "authority", "scheme", "path"],
    headers_priority=lkrequest.HeadersPriority(0, 255, True),
)

# Custom TCP Fingerprint
tcp = lkrequest.TcpFingerprint(window_size=65535, mss=1460, window_scale=8, ttl=128)

# Combine them
client = lkrequest.Client(tls_profile=tls, h2_profile=h2, tcp_fingerprint=tcp)

# Serialize / deserialize
json_str = tls.to_json()
restored = lkrequest.TlsProfile.from_json(json_str)

# Fingerprint randomization
randomized = client.randomize_fingerprint(shuffle_extensions=True)

# Randomization strategy (Randomize): pass to Client(randomize=...)
# Tier 1 (always available): permute TLS extension order per connection, drift JA3 but stay a real browser
client = lkrequest.Client(tls_profile="chrome_146",
                          randomize=lkrequest.Randomize.extension_order())

# Tier 3a/3b (requires the synthetic-fp feature): synthesize a cross-layer unique identity per session
#   from lkrequest import Randomize, Layers
#   client = lkrequest.Client(tls_profile="chrome_146",
#                             randomize=Randomize.recombine())                 # all layers
#   client = lkrequest.Client(tls_profile="chrome_146",
#                             randomize=Randomize.recombine_layers(Layers.TLS | Layers.H2))

Fingerprint consistency validation

result = lkrequest.validate_fingerprint_consistency(
    tls_profile="chrome_144", h2_profile="chrome_144", tcp_fingerprint="chrome_win",
)
print(result["valid"])     # True
print(result["warnings"])  # []

Timeout configuration

# Client-level timeouts
client = lkrequest.Client(
    dns_timeout=5.0,
    tcp_connect_timeout=10.0,
    tls_handshake_timeout=10.0,
    ttfb_timeout=15.0,
    total_timeout=30.0,
)

# Request-level timeout override
resp = session.get("https://example.com", timeout=5.0)

# TimeoutConfig object
tc = lkrequest.TimeoutConfig(total=30.0, tcp_connect=10.0)

# ResourceLimits
rl = lkrequest.ResourceLimits(max_response_body_size=10*1024*1024, max_connections_per_session=16)

AcceptEncoding control

ae = lkrequest.AcceptEncoding

# Session level
session = client.session(accept_encoding=ae.GZIP | ae.BR)

# Request level
resp = session.get(url, accept_encoding=ae.GZIP)

# Disable automatic decompression
resp = session.get(url, no_decompress=True)

Header / Cookie order

header_order (header send order) and cookie_order (ordering within the Cookie header) can be set at three levels, with lower levels overriding higher ones:

# Client level (applies to all sessions of this client)
client = lkrequest.Client(
    header_order=["host", "user-agent", "accept", "accept-encoding", "cookie"],
    cookie_order=["session_id", "csrf_token"],
)

# Session level (overrides client level)
session = client.session(
    header_order=["host", "user-agent", "accept"],
    cookie_order=["session_id", "csrf_token"],
)

# Request level (overrides session and client levels)
resp = await session.get(
    url,
    header_order=["host", "user-agent", "accept"],
    cookie_order=["session_id", "csrf_token"],
)

Ordered / duplicate headers, params, data

headers, params, and data accept either a dict (preserves insertion order) or a list[tuple[str, str]] (preserves order and allows duplicate keys, for repeated params like ?tag=a&tag=b):

# Duplicate query params: ?tag=a&tag=b
resp = session.get(url, params=[("tag", "a"), ("tag", "b")])

# Duplicate form fields
resp = session.post(url, data=[("k", "1"), ("k", "2")])

# Explicit header ordering (a dict is also sent in insertion order)
resp = session.get(url, headers=[("user-agent", "..."), ("accept", "*/*")])

base_url, generic request(), connection pool cleanup

# base_url: relative paths are auto-joined; passing an absolute URL overrides base_url
session = client.session(base_url="https://api.example.com")
resp = await session.get("/v1/users", params={"page": "1"})

# Generic request(): specify the method explicitly (any case)
resp = await session.request("PATCH", "/v1/users/1", json={"name": "x"})

# Only allow redirects to HTTPS; clear the connection pool
session = client.session(https_only=True)
session.pool_clear()

HSTS / http→https upgrade policy

Browsers upgrade http:// requests to https:// before connecting for known HSTS (or preloaded) hosts. The default (no hsts) is a stateless "first visit" that performs no upgrade. Host matching is suffix-based (equivalent to HSTS includeSubDomains).

# Dynamically learn HSTS hosts within the session from response Strict-Transport-Security headers (like a long-running browser)
session = client.session(hsts=lkrequest.Hsts.dynamic())

# Seed known HSTS hosts (e.g. replay state learned in a previous session)
session = client.session(hsts=lkrequest.Hsts.dynamic(seed=["example.com"]))

# Fixed allowlist; preloaded_tlds=True also upgrades Chrome's fully-preloaded gTLDs (.dev/.app/…)
session = client.session(hsts=lkrequest.Hsts.static_(["example.com"], preloaded_tlds=True))

Request-level protocol override

# Each request can override session-level protocol policy / intent, plus HTTP/3-specific header order
resp = await session.get(
    url,
    protocol_policy=lkrequest.ProtocolPolicy.chrome_conservative(),
    http_intent=lkrequest.HttpIntent.H2Only,
    h3_header_order=["host", "user-agent", "accept"],
)

h3_header_order can also be set on Client(...) and session(...) (same three levels as header_order).

Prometheus metrics

collector = lkrequest.enable_metrics()

# View stats after making requests
stats = collector.snapshot()
print(collector.prometheus_text())
collector.reset()

Request diagnostics & runtime metrics

response.diagnostics returns per-request phase timings (no OpenTelemetry dependency, collected directly by the Rust engine):

resp = await session.get("https://example.com")
print(resp.diagnostics)
# {'dns_ms': 3, 'tcp_ms': 0, 'tls_ms': 423, 'ttfb_ms': 193, 'total_ms': 621,
#  'remote_addr': '93.184.216.34:443', 'protocol': 'h2', 'cipher_suite': '0x1301'}
# dns_ms/tcp_ms/tls_ms are None on a reused connection (that phase did not occur)

metrics_snapshot() pulls process-level runtime counters for you to feed into your own Prometheus / OTel pipeline:

snap = lkrequest.metrics_snapshot()
# {'bytes_in': ..., 'bytes_out': ..., 'requests_total': ...,
#  'requests_failed': ..., 'active_connections': ..., 'total_connections': ...}

Counters are all 0 by default; only building with the telemetry feature (maturin build --features telemetry) enables transport-layer byte counting.

Certificate management

client = lkrequest.Client(ca_cert="/path/to/ca.pem")       # file path
client = lkrequest.Client(ca_cert_pem=pem_bytes)            # PEM bytes
client = lkrequest.Client(ca_cert_der=der_bytes)            # DER bytes
client = lkrequest.Client(verify=False)                      # disable verification (dev only)
client = lkrequest.Client(use_native_certs=True)             # system certificates
client = lkrequest.Client(ech_config=ech_bytes)              # ECH support

Zero-copy Response

resp = session.get("https://example.com")

mv = memoryview(resp)       # zero-copy access to body
text = resp.text()          # cached after first decode
data = resp.json()          # cached after first parse

Logging

lkrequest.set_log_level("info")                          # global level
lkrequest.set_log_level("lkrequest=debug,lktls=trace")   # module filter
lkrequest.set_log_level("info", format="compact")        # log format
lkrequest.set_log_level("off")                           # disable

Error handling

try:
    resp = session.get(url, timeout=10.0)
    resp.error_for_status()
except lkrequest.LkTimeoutError:
    print("timed out")
except lkrequest.LkConnectionError:
    print("connection failed")
except lkrequest.TlsError:
    print("TLS error")
except lkrequest.HttpStatusError as e:
    print(f"HTTP error: {e}")
except lkrequest.TooManyRedirectsError:
    print("too many redirects")
except lkrequest.RequestError as e:
    print(f"request error: {e}")

Available Browser Fingerprints

TCP fingerprints are OS-specific: chrome_win / chrome_linux / chrome_macos / firefox_win / firefox_linux / firefox_macos / safari

API Reference

Client

Session

All HTTP methods (get/post/put/delete/head/patch/options) support:

Response

HeaderMap

Fingerprint configuration

Module functions

Exceptions

Examples

The examples/ directory contains complete examples for every feature:

Run an example:

maturin develop
python examples/basic_requests.py

Testing

Using venv and pip:

# After completing the development setup above, run all tests
just test

# Or manually
python -m pip install -e ".[test]"
python -m pytest tests/ -v

# Run only the unit tests that don't need network
python -m pytest tests/ -v -k "not httpbin and not echo"

Using uv:

uv sync --extra test --extra lint
just test
# Or:
uv run pytest tests/ -v

Benchmark

maturin develop --release
python benchmarks/bench_latency.py     # single-request latency
python benchmarks/bench_throughput.py   # concurrent throughput
python benchmarks/bench_memory.py      # memory usage
python benchmarks/verify_fingerprint.py # TLS fingerprint verification