HTTP client

English | Русский

• HTTP client
  • Installation
  • Skills for AI agents
  • Usage
    • Example:
      • Base client example
      • Custom client example
      • Asynchronous client example
    • Configuration
  • Available options
    • EventHooks
    • Decorating the send method
    • Streaming requests
      • The stream method
      • The stream_request method
      • Decorating the stream_request method
    • The request method
    • Timeouts
    • Authorization
    • Request identification
    • Tracing
    • Logging
    • Closing connections
    • TLS connection
    • Default headers
  • Development

Installation

Requires Python 3.13 or later.

pip install cloudru-qa-http-client

Import the library as the http_client module.

Skills for AI agents

The package includes the http-client-usage skill with examples and usage rules for AI agents. Install the library in your project environment, then run this from its root (requires uv):

uvx library-skills install --skill http-client-usage --yes

For Claude Code, add --claude. The instructions are shipped with the library version.

Usage

The library provides HttpClient and AsyncHttpClient. Use them directly or subclass them to create custom clients.

All options (logging, request identification, tracing, default headers, and others) work the same for both clients. Synchronous event hooks are automatically wrapped as async when used with AsyncHttpClient.

Example:

Base client example

import logging

from http_client import HttpClient, options
from httpx import BasicAuth

with HttpClient(
    "https://api.example.com",
    options.with_auth(BasicAuth("demo", "example-password")),
    options.with_logger(logging.getLogger(__name__)),
) as client:
    response = client.post("/test_method", json={"test": True})
    response.raise_for_status()

Custom client example

import logging

from http_client import HttpClient, options
from httpx import Response


class CustomHttpClient(HttpClient):
    def test_method(self, data: dict[str, object]) -> Response:
        """Send data to a service method.

        :param data: JSON request body.
        :return: service response.
        """
        request = self.client.build_request("POST", "/test_method", json=data)
        return self.send(request)


with CustomHttpClient(
    "https://api.example.com",
    options.with_logger(logging.getLogger(__name__)),
    options.with_autogenerated_request_id(),
    options.with_autogenerated_traceparent(),
) as client:
    response = client.test_method({"test": True})
    response.raise_for_status()

Asynchronous client example

All standard options, including synchronous event hooks, work unchanged with AsyncHttpClient:

import logging

from http_client import AsyncHttpClient, options

async with AsyncHttpClient(
    "https://api.example.com",
    options.with_logger(logging.getLogger(__name__)),
    options.with_autogenerated_request_id(),
) as client:
    response = await client.post("/test_method", json={"test": True})
    response.raise_for_status()

Configuration

Configure the client with ClientOption options. Each option receives the actual client instance and runs in its constructor. Register hooks and decorators with add_request_hook, add_response_hook, add_send_decorator, and add_stream_decorator:

from http_client import BaseHttpClient, ClientOption
from httpx import Request


def with_region(region: str) -> ClientOption:
    """Add a region to outgoing request headers.

    :param region: region identifier.
    :return: client option.
    """

    def add_region(request: Request) -> None:
        request.headers["X-Region"] = region

    def option(client: BaseHttpClient) -> None:
        client.add_request_hook(add_region, prepend=True)

    return option

Registration is available while constructor options run. After client creation, its transport is already configured.

Available options

The options module provides options for configuring the HTTP client. Combine them when creating a client instance.

EventHooks

Event hook options let you process requests and responses:

• with_request_hook — adds a function to process requests before sending
• with_response_hook — adds a function to process responses after receiving

These options provide a flexible way to handle each client request and/or response.

Usage example:

import logging

from httpx import Request, Response

from http_client import HttpClient, options


def my_request_hook(request: Request) -> None:
    # process the request before sending it
    request.headers["X-Custom-Header"] = "Custom Value"


def my_response_hook(response: Response) -> None:
    # process the response after receiving it
    logging.getLogger(__name__).info("Response status code: %s", response.status_code)


client = HttpClient(
    "http://example.com",
    options.with_request_hook(my_request_hook),
    options.with_response_hook(my_response_hook),
)

With AsyncHttpClient, synchronous hooks are automatically wrapped as async. You can also pass native async hooks:

from http_client import AsyncHttpClient, options
from httpx import Request


async def my_async_request_hook(request: Request) -> None:
    request.headers["X-Async"] = "true"


client = AsyncHttpClient(
    "http://example.com",
    options.with_request_hook(my_async_request_hook),
)

See the httpx documentation for more about hooks.

Decorating the send method

The HTTP client supports decorating its send method with with_send_decorator. This lets you add behavior before sending the HTTP request and after receiving the response.

with_send_decorator applies the decorator pattern to send, letting you:

• Preprocess the request
• Postprocess the response
• Implement retries, logging, caching, monitoring, and other features

Example of a request logging decorator:

from functools import wraps
from http_client import HttpClient, options
from http_client.types import SendOptions, SyncSendMethod
from typing import Unpack
import logging
import httpx


def logging_decorator(func: SyncSendMethod) -> SyncSendMethod:
    @wraps(func)
    def wrapper(request: httpx.Request, **kwargs: Unpack[SendOptions]) -> httpx.Response:
        logging.getLogger(__name__).info("Sending request: %s %s", request.method, request.url)
        response = func(request, **kwargs)
        logging.getLogger(__name__).info("Received response: %s", response.status_code)
        return response

    return wrapper


client = HttpClient("https://example.com", options.with_send_decorator(logging_decorator))

For the asynchronous HTTP client, the decorator must be asynchronous:

from functools import wraps
from http_client import AsyncHttpClient, options
from http_client.types import AsyncSendMethod, SendOptions
from typing import Unpack
import logging
import httpx


def async_logging_decorator(func: AsyncSendMethod) -> AsyncSendMethod:
    @wraps(func)
    async def wrapper(request: httpx.Request, **kwargs: Unpack[SendOptions]) -> httpx.Response:
        logging.getLogger(__name__).info("Sending request: %s %s", request.method, request.url)
        response = await func(request, **kwargs)
        logging.getLogger(__name__).info("Received response: %s", response.status_code)
        return response

    return wrapper


client = AsyncHttpClient("https://example.com", options.with_send_decorator(async_logging_decorator))

You can apply several decorators. Each new decorator wraps the previous one, so the last one added is outermost and runs first:

client = HttpClient(
    "https://example.com",
    options.with_send_decorator(decorator1),  # applied first, becomes innermost
    options.with_send_decorator(decorator2),  # applied second, becomes outermost
)

Request calls run in this order: decorator2 → decorator1 → send. The response passes through them in reverse: decorator1 finishes first, then decorator2.

Streaming requests

The HTTP client supports streaming requests in two ways:

The stream method

stream(method, url, **kwargs) is fully compatible with the httpx API. It takes an HTTP method and URL, builds a request, and returns a context manager:

from http_client import HttpClient, options

client = HttpClient(
    "https://example.com",
    options.with_auth(auth),
    options.with_logger(logger),
)

# Streaming download (httpx contract)
with client.stream("GET", "/large-file.zip") as response:
    response.raise_for_status()

    with open("file.zip", "wb") as f:
        for chunk in response.iter_bytes(chunk_size=8192):
            f.write(chunk)

Asynchronous client example:

from http_client import AsyncHttpClient, options

client = AsyncHttpClient(
    "https://example.com",
    options.with_auth(auth),
    options.with_logger(logger),
)

async with client.stream("GET", "/stream-data") as response:
    response.raise_for_status()

    async for line in response.aiter_lines():
        process_line(line)

The stream_request method

stream_request(request, **kwargs) takes a ready-made httpx.Request and supports all client options, including decorators:

request = client.client.build_request("GET", "/large-file.zip")

with client.stream_request(request) as response:
    for chunk in response.iter_bytes(chunk_size=8192):
        f.write(chunk)

Decorating the stream_request method

Like send, stream_request can be decorated with with_stream_decorator:

from functools import wraps
from contextlib import contextmanager
from http_client import HttpClient, options
from http_client.types import StreamOptions, SyncStreamRequestMethod
from collections.abc import Iterator
from typing import Unpack
import httpx
import logging


def stream_logging_decorator(func: SyncStreamRequestMethod) -> SyncStreamRequestMethod:
    @wraps(func)
    @contextmanager
    def wrapper(request: httpx.Request, **kwargs: Unpack[StreamOptions]) -> Iterator[httpx.Response]:
        logging.getLogger(__name__).info("Starting stream: %s %s", request.method, request.url)
        with func(request, **kwargs) as response:
            logging.getLogger(__name__).info("Stream status: %s", response.status_code)
            yield response
        logging.getLogger(__name__).info("Stream finished")

    return wrapper


client = HttpClient("https://example.com", options.with_stream_decorator(stream_logging_decorator))

Important: When logging streaming requests, the response body appears as <streamed content> because it has not yet been read when the response hook runs.

The request method

request(method, url, **kwargs) sends HTTP requests with any method:

response = client.request("PATCH", "/api/resource", json={"field": "value"})

Timeouts

with_timeout sets different types of HTTP request timeouts:

from httpx import Timeout
from http_client import HttpClient, options

# Set different timeout types
timeout = Timeout(
    connect=5.0,  # time to establish a connection
    read=10.0,  # time to read the response
    write=5.0,  # time to write the request
    pool=1.0,  # time to wait for a free connection in the pool
)

client = HttpClient("https://example.com", options.with_timeout(timeout))

Authorization

with_auth configures HTTP request authorization. All httpx authorization mechanisms are supported. You can also implement your own; see the httpx documentation.

from httpx import BasicAuth, DigestAuth
from http_client import HttpClient, options

# Basic authorization
client = HttpClient("https://example.com", options.with_auth(BasicAuth(username="user", password="password")))

# Digest authorization
client = HttpClient("https://example.com", options.with_auth(DigestAuth(username="user", password="password")))

# Custom Bearer token authorization
client = HttpClient(
    "https://example.com", options.with_auth(lambda r: r.headers.update({"Authorization": f"Bearer {token}"}))
)

Request identification

with_autogenerated_request_id automatically adds a unique identifier to each request's X-Request-ID header:

from http_client import HttpClient, options

# Automatically generate X-Request-ID
client = HttpClient("https://example.com", options.with_autogenerated_request_id())

This is useful for tracing requests across microservices and for debugging.

Tracing

with_autogenerated_traceparent automatically adds a traceparent tracing header compatible with W3C Trace Context:

from http_client import HttpClient, options

# Add a traceparent header to every request
client = HttpClient("https://example.com", options.with_autogenerated_traceparent())

This lets you track requests across multiple services using different monitoring systems.

Logging

with_logger configures HTTP request and response logging. Set shared defaults with a LoggingConfig object:

from http_client import configure_logging
from http_client.config import LoggingConfig

configure_logging(options=LoggingConfig(json_indent=2, use_line_breaks=True))

Pass client-specific settings to with_logger:

import logging
from http_client import HttpClient, options

# Configure a custom logger
logger = logging.getLogger("http_client")

client = HttpClient(
    "https://example.com",
    options.with_logger(
        logger,
        sanitize_keys=["authorization", "password"],  # hide secret data
    ),
)

sanitize_keys hides sensitive information in logs, such as authorization tokens and passwords.

Common secrets are masked by default, even without sanitize_keys: authorization, proxy-authorization, cookie, set-cookie, x-api-key, api-key, api_key, access_token, refresh_token, password, secret, token. Keys from sanitize_keys extend this list.

Disable the default protection with an explicit flag, for example when debugging in a private environment:

options.with_logger(logger, use_default_sanitize_keys=False)

⚠️ Masking applies only to access log records written by this library. httpx exceptions may contain the full URL, including query parameters, so do not pass secrets in URLs.

Closing connections

The connection pool is created in the client constructor, so close the client. A context manager is the simplest way:

with HttpClient("https://example.com") as client:
    client.get("/api/resource")

async with AsyncHttpClient("https://example.com") as client:
    await client.get("/api/resource")

For a long-lived client created once per application, use explicit methods:

client.close()  # HttpClient
await client.aclose()  # AsyncHttpClient

A closed client cannot be reopened: like httpx, this raises RuntimeError.

TLS connection

The library provides two options to configure TLS connections. Both work the same with HttpClient and AsyncHttpClient.

1. with_verify_context — sets the TLS verification context:

import ssl
from http_client import HttpClient, options

# Create an SSL context
ssl_context = ssl.create_default_context()
ssl_context.load_verify_locations("/path/to/ca-certificates.crt")

# Set the SSL context
client = HttpClient("https://example.com", options.with_verify_context(verify=ssl_context))

2. use_truststore — uses the system certificate store for verification:

from http_client import HttpClient, options

# Use the system certificate store
client = HttpClient("https://example.com", options.use_truststore())

To use use_truststore, install the extra dependency:

pip install "cloudru-qa-http-client[truststore]"

Default headers

with_default_headers sets default headers sent with all client requests.

from http_client import HttpClient, options

headers = {
    "X-My-Custom-Header": "Hello, world!",
    "X-Another-Header": "Another value",
}

client = HttpClient(
    "https://example.com",
    options.with_default_headers(headers),
)

Development

The library is developed in the qa-libraries monorepo alongside the other cloudru-qa-* libraries. You need uv 0.11.14 or later, GNU Make, and checkmake. Run commands from the repository root:

make sync                     # install dependencies for all libraries from uv.lock
make test-http-client         # tests for this library
make typecheck-http-client    # mypy for this library
make check                    # formatting, ruff, mypy, and tests for the entire workspace

⚠️ Keep the library lightweight and general-purpose. Discuss new dependencies only when you are sure all library users need them.

Tests run on Python 3.13 and 3.14. Create a separate environment for another version: make sync test-http-client PYTHON_VERSION=3.14.

Demo scenarios with real requests to local go-httpbin are in examples/httpbin. Docker must be running; run commands from packages/http-client:

docker compose up -d
uv run --no-sync python -m examples.httpbin
docker compose down

Publish a release with the git tag http-client/v<version>. The tag version must match the version in packages/http-client/pyproject.toml.