Protocol Interfaces¶

httptap uses Protocol classes (PEP 544) for structural subtyping, so you can supply custom implementations without inheriting from any base class.

Why protocols?¶

Duck typing with type safety — type checkers verify your implementation
No inheritance required — just implement the methods
Clear contracts — explicit interface definitions
Easy testing — simple to mock and substitute

The interface contracts below are rendered from source. Each is followed by a worked custom implementation you can adapt.

DNSResolver¶

DNSResolver ¶

Bases: Protocol

Protocol for DNS resolution implementations.

This protocol defines the interface for DNS resolvers that can translate hostnames to IP addresses with timing measurements.

A resolver may also define resolve_all(host, port, timeout) returning every address as (ip, family) pairs; httptap then tries them in order and falls back to the next one when a connection fails. It is used only when declared on the class that defines resolve or a subclass of it, so a subclass that overrides just resolve keeps working.

示例：

>>> class CustomDNSResolver:
...     def resolve(
...         self, host: str, port: int, timeout: float
...     ) -> tuple[str, str, float]:
...         # Custom DNS resolution logic
...         return "93.184.216.34", "IPv4", 12.5

方法：

resolve –

Resolve hostname to IP address with timing.

resolve ¶

resolve(
    host: str, port: int, timeout: float
) -> tuple[str, str, float]

Resolve hostname to IP address with timing.

参数：

host (str) –

Hostname to resolve (e.g., "example.com").
port (int) –

Port number for the connection.
timeout (float) –

Maximum time to wait for DNS resolution in seconds.

返回：

str –

Tuple of (ip_address, ip_family, resolution_time_ms).
str –

ip_family should be one of: 'IPv4', 'IPv6', or 'AF_'.

引发：

DNSResolutionError –

If the hostname cannot be resolved. httptap reports it as a network error (exit code 75); any other exception is treated as an internal error (exit code 70).

Note

Implementations should try multiple resolution methods (e.g., system resolver, custom nameservers) before failing.

源代码位于： httptap/interfaces.py

def resolve(self, host: str, port: int, timeout: float) -> tuple[str, str, float]:
    """Resolve hostname to IP address with timing.

    Args:
        host: Hostname to resolve (e.g., "example.com").
        port: Port number for the connection.
        timeout: Maximum time to wait for DNS resolution in seconds.

    Returns:
        Tuple of (ip_address, ip_family, resolution_time_ms).
        ip_family should be one of: 'IPv4', 'IPv6', or 'AF_<num>'.

    Raises:
        DNSResolutionError: If the hostname cannot be resolved. httptap
            reports it as a network error (exit code 75); any other
            exception is treated as an internal error (exit code 70).

    Note:
        Implementations should try multiple resolution methods
        (e.g., system resolver, custom nameservers) before failing.

    """

httptap dials the resolved IP address directly while keeping the original hostname for the Host header and TLS SNI. IPv6 addresses are bracketed automatically; implementations only need to return a valid (ip, family, duration_ms) tuple. family is "IPv4", "IPv6", or "AF_<num>" for other address families. The reported dns_ms is measured by httptap around the resolver call, so duration_ms is informational. Internationalized hostnames are passed in their IDNA 2008 A-label form (xn--…), the same name httptap sends in Host and SNI.

Raise DNSResolutionError (exported from httptap) when a name cannot be resolved. httptap records it as a failed step with a network error (exit code 75). Any other exception is treated as an internal error (exit code 70).

Example implementation¶

import socket
import time

from httptap import DNSResolutionError, HTTPTapAnalyzer


class CustomDNSResolver:
    def resolve(self, host: str, port: int, timeout: float) -> tuple[str, str, float]:
        start = time.perf_counter()
        try:
            addr_info = socket.getaddrinfo(host, port, socket.AF_UNSPEC, socket.SOCK_STREAM)
        except socket.gaierror as e:
            raise DNSResolutionError(f"DNS resolution failed for {host}: {e}") from e
        ip_address = addr_info[0][4][0]
        family = "IPv6" if ":" in ip_address else "IPv4"
        duration_ms = (time.perf_counter() - start) * 1000
        return ip_address, family, duration_ms


analyzer = HTTPTapAnalyzer(dns_resolver=CustomDNSResolver())

Address fallback with `resolve_all()`¶

A resolver may also implement the optional resolve_all() method. It is not part of the DNSResolver protocol, but httptap uses it when it is available:

def resolve_all(self, host: str, port: int, timeout: float) -> tuple[list[tuple[str, str]], float]: ...

It returns every usable address as (ip, family) pairs in the order to try, plus the resolution time in milliseconds. httptap connects to the addresses in that order and moves to the next one when a connection fails or times out; the connect timeout is split across the remaining addresses, and TLS errors are not retried. Through a local-DNS socks5:// proxy, httptap moves on when the proxy reports that it cannot connect to an address; failures to reach or authenticate with the proxy itself are not retried. Time spent on addresses that failed is included in connect_ms and total_ms, as curl's time_connect does. An empty list is reported as a DNS failure. SystemDNSResolver implements both methods.

resolve_all() is used only when the class that defines it is the class that defines resolve() or a subclass of it. A subclass that overrides only resolve() therefore keeps working as written: httptap calls the overridden resolve() and connects to that single address instead of using the inherited resolve_all(). Override resolve_all() as well to keep address fallback:

from httptap import SystemDNSResolver

PINNED = {"api.example.com": "203.0.113.10"}


class PinnedResolver(SystemDNSResolver):
    def resolve(self, host: str, port: int, timeout: float) -> tuple[str, str, float]:
        if host in PINNED:
            return PINNED[host], "IPv4", 0.0
        return super().resolve(host, port, timeout)

    def resolve_all(self, host: str, port: int, timeout: float) -> tuple[list[tuple[str, str]], float]:
        if host in PINNED:
            return [(PINNED[host], "IPv4")], 0.0
        return super().resolve_all(host, port, timeout)

TLSInspector¶

TLSInspector ¶

Bases: Protocol

Protocol for TLS/SSL inspection implementations.

This protocol defines the interface for inspecting TLS connections to extract certificate information and connection metadata.

示例：

>>> class CustomTLSInspector:
...     def inspect(
...         self,
...         host: str,
...         port: int,
...         timeout: float,
...     ) -> NetworkInfo:
...         # Custom TLS inspection logic
...         return NetworkInfo(tls_version="TLSv1.3")

方法：

inspect –

Inspect TLS connection and extract metadata.

inspect ¶

inspect(
    host: str, port: int, timeout: float
) -> NetworkInfo

Inspect TLS connection and extract metadata.

参数：

host (str) –

Hostname to connect to.
port (int) –

Port number (typically 443 for HTTPS).
timeout (float) –

Connection timeout in seconds.

返回：

NetworkInfo –

NetworkInfo object with TLS version, cipher, and certificate data.

Note

Implementations should handle connection failures gracefully and return partial data when possible (e.g., TLS version without certificate details if handshake succeeds but cert extraction fails).

源代码位于： httptap/interfaces.py

def inspect(
    self,
    host: str,
    port: int,
    timeout: float,
) -> NetworkInfo:
    """Inspect TLS connection and extract metadata.

    Args:
        host: Hostname to connect to.
        port: Port number (typically 443 for HTTPS).
        timeout: Connection timeout in seconds.

    Returns:
        NetworkInfo object with TLS version, cipher, and certificate data.

    Note:
        Implementations should handle connection failures gracefully
        and return partial data when possible (e.g., TLS version without
        certificate details if handshake succeeds but cert extraction fails).

    """

httptap reads TLS and certificate details from the live connection that served the response. A custom inspector is only a fallback: it is called for HTTPS requests when that connection exposes no TLS data and no proxy is in use, receives the hostname, port, and remaining timeout, and opens its own connection. Raise TLSInspectionError (exported from httptap) when inspection fails; the step is then reported without TLS details. The diagnostic probe that runs after a certificate verification failure always uses the built-in SocketTLSInspector, not a custom inspector.

Example implementation¶

import ssl
import socket
import time
from datetime import datetime
from httptap.models import NetworkInfo


class CustomTLSInspector:
    def inspect(self, host: str, port: int, timeout: float) -> NetworkInfo:
        context = ssl.create_default_context()
        with socket.create_connection((host, port), timeout=timeout) as sock:
            with context.wrap_socket(sock, server_hostname=host) as ssock:
                version = ssock.version()
                cipher = ssock.cipher()[0]
                cert = ssock.getpeercert()
                cert_cn = dict(x[0] for x in cert["subject"])["commonName"]
                not_after = datetime.strptime(cert["notAfter"], "%b %d %H:%M:%S %Y %Z")
                days_left = (not_after - datetime.now()).days

        return NetworkInfo(
            tls_version=version,
            tls_cipher=cipher,
            cert_cn=cert_cn,
            cert_days_left=days_left,
        )


from httptap import HTTPTapAnalyzer

analyzer = HTTPTapAnalyzer(tls_inspector=CustomTLSInspector())

TimingCollector¶

A new collector instance is created for each request in the chain, so pass the class (a factory), not an instance.

TimingCollector ¶

Bases: Protocol

Protocol for collecting timing metrics during HTTP requests.

This protocol defines the interface for components that measure and track timing information throughout request execution phases.

示例：

>>> class CustomTimingCollector:
...     def mark_dns_start(self) -> None:
...         self._dns_start = time.time()
...     def get_metrics(self) -> TimingMetrics:
...         return TimingMetrics(dns_ms=self._dns_ms)

方法：

mark_dns_start –

Mark the start of DNS resolution phase.
mark_dns_end –

Mark the end of DNS resolution phase.
mark_request_start –

Mark the start of HTTP request phase.
mark_ttfb –

Mark the time to first byte (headers received).
mark_request_end –

Mark the end of HTTP request (body fully received).
get_metrics –

Calculate and return timing metrics.

mark_dns_start ¶

mark_dns_start() -> None

Mark the start of DNS resolution phase.

源代码位于： httptap/interfaces.py

def mark_dns_start(self) -> None:
    """Mark the start of DNS resolution phase."""

mark_dns_end ¶

mark_dns_end() -> None

Mark the end of DNS resolution phase.

源代码位于： httptap/interfaces.py

def mark_dns_end(self) -> None:
    """Mark the end of DNS resolution phase."""

mark_request_start ¶

mark_request_start() -> None

Mark the start of HTTP request phase.

源代码位于： httptap/interfaces.py

def mark_request_start(self) -> None:
    """Mark the start of HTTP request phase."""

mark_ttfb ¶

mark_ttfb() -> None

Mark the time to first byte (headers received).

源代码位于： httptap/interfaces.py

def mark_ttfb(self) -> None:
    """Mark the time to first byte (headers received)."""

mark_request_end ¶

mark_request_end() -> None

Mark the end of HTTP request (body fully received).

源代码位于： httptap/interfaces.py

def mark_request_end(self) -> None:
    """Mark the end of HTTP request (body fully received)."""

get_metrics ¶

get_metrics() -> TimingMetrics

Calculate and return timing metrics.

返回：

TimingMetrics –

TimingMetrics with all phase durations calculated.

Note

Should calculate derived metrics (wait_ms, xfer_ms) automatically.

源代码位于： httptap/interfaces.py

def get_metrics(self) -> TimingMetrics:
    """Calculate and return timing metrics.

    Returns:
        TimingMetrics with all phase durations calculated.

    Note:
        Should calculate derived metrics (wait_ms, xfer_ms) automatically.

    """

Example implementation¶

import time
from httptap.models import TimingMetrics


class CustomTimingCollector:
    def __init__(self) -> None:
        self._dns_start = 0.0
        self._dns_end = 0.0
        self._request_start = 0.0
        self._ttfb = 0.0
        self._request_end = 0.0

    def mark_dns_start(self) -> None:
        self._dns_start = time.perf_counter()

    def mark_dns_end(self) -> None:
        self._dns_end = time.perf_counter()

    def mark_request_start(self) -> None:
        self._request_start = time.perf_counter()

    def mark_ttfb(self) -> None:
        self._ttfb = time.perf_counter()

    def mark_request_end(self) -> None:
        self._request_end = time.perf_counter()

    def get_metrics(self) -> TimingMetrics:
        dns_ms = (self._dns_end - self._dns_start) * 1000
        ttfb_ms = (self._ttfb - self._dns_start) * 1000
        total_ms = (self._request_end - self._dns_start) * 1000
        metrics = TimingMetrics(dns_ms=dns_ms, ttfb_ms=ttfb_ms, total_ms=total_ms)
        metrics.calculate_derived()
        return metrics


# Pass the class (not an instance) as the factory:
from httptap import HTTPTapAnalyzer

analyzer = HTTPTapAnalyzer(timing_collector_factory=CustomTimingCollector)

Visualizer¶

Visualizer ¶

Bases: Protocol

Renderable component capable of visualising a single step.

This protocol defines the interface for visualizers that can render HTTP request analysis steps in various formats (waterfall, ASCII, etc.).

示例：

>>> class CustomVisualizer:
...     def render(self, step: StepMetrics) -> None:
...         print(f"Step {step.step_number}: {step.timing.total_ms}ms")

方法：

render –

Render a visualisation for the provided HTTP step.

render ¶

render(step: StepMetrics) -> None

Render a visualisation for the provided HTTP step.

参数：

step (StepMetrics) –

Step metrics containing timing, network, and response data.

Note

Implementations should handle errors gracefully and avoid raising exceptions to prevent disrupting the analysis output.

源代码位于： httptap/interfaces.py

def render(self, step: StepMetrics) -> None:
    """Render a visualisation for the provided HTTP step.

    Args:
        step: Step metrics containing timing, network, and response data.

    Note:
        Implementations should handle errors gracefully and avoid raising
        exceptions to prevent disrupting the analysis output.

    """

Example implementation¶

from httptap.models import StepMetrics


class SimpleVisualizer:
    def render(self, step: StepMetrics) -> None:
        print(f"Step {step.step_number}: {step.url}")
        print(f"  Status: {step.response.status}")
        print(f"    DNS:     {step.timing.dns_ms:8.2f}ms")
        print(f"    Connect: {step.timing.connect_ms:8.2f}ms")
        print(f"    TLS:     {step.timing.tls_ms:8.2f}ms")
        print(f"    TTFB:    {step.timing.ttfb_ms:8.2f}ms")
        print(f"    Total:   {step.timing.total_ms:8.2f}ms")


from httptap import HTTPTapAnalyzer

analyzer = HTTPTapAnalyzer()
for step in analyzer.analyze_url("https://httpbin.io"):
    SimpleVisualizer().render(step)

Exporter¶

Exporter ¶

Bases: Protocol

Component responsible for exporting analysis output.

This protocol defines the interface for exporters that can persist analysis results in various formats (JSON, CSV, HTML, etc.).

示例：

>>> class CSVExporter:
...     def export(
...         self,
...         steps: Sequence[StepMetrics],
...         initial_url: str,
...         output_path: str,
...     ) -> None:
...         # Write CSV file
...         pass

方法：

export –

Persist the collected steps using the chosen representation.

export ¶

export(
    steps: Sequence[StepMetrics],
    initial_url: str,
    output_path: str,
    *,
    slo_result: SLOResult | None = None,
) -> None

Persist the collected steps using the chosen representation.

参数：

steps (Sequence[StepMetrics]) –

Sequence of step metrics to export.
initial_url (str) –

The initial URL that was analyzed.
output_path (str) –

Path to output file where results should be written.
slo_result (SLOResult | None, 默认： None ) –

Optional SLO evaluation result that concrete exporters may embed alongside the step data.

引发：

IOError –

If file cannot be written or path is invalid.

Note

Implementations should create parent directories if they don't exist.

源代码位于： httptap/interfaces.py

def export(
    self,
    steps: Sequence[StepMetrics],
    initial_url: str,
    output_path: str,
    *,
    slo_result: SLOResult | None = None,
) -> None:
    """Persist the collected steps using the chosen representation.

    Args:
        steps: Sequence of step metrics to export.
        initial_url: The initial URL that was analyzed.
        output_path: Path to output file where results should be written.
        slo_result: Optional SLO evaluation result that concrete
            exporters may embed alongside the step data.

    Raises:
        IOError: If file cannot be written or path is invalid.

    Note:
        Implementations should create parent directories if they don't exist.

    """

Concrete exporters may embed an optional SLO evaluation via the keyword-only slo_result argument; see the built-in JSONExporter.

Example implementation¶

import yaml
from collections.abc import Sequence
from httptap.models import StepMetrics
from httptap.slo import SLOResult


class YAMLExporter:
    def export(
        self,
        steps: Sequence[StepMetrics],
        initial_url: str,
        output_path: str,
        *,
        slo_result: SLOResult | None = None,
    ) -> None:
        data = {
            "initial_url": initial_url,
            "total_steps": len(steps),
            "steps": [
                {
                    "url": step.url,
                    "status": step.response.status,
                    "timing": step.timing.to_dict(),
                    "network": step.network.to_dict(),
                }
                for step in steps
            ],
        }
        if slo_result is not None:
            data["slo"] = slo_result.to_dict()
        with open(output_path, "w") as f:
            yaml.dump(data, f, default_flow_style=False)


from httptap import HTTPTapAnalyzer

analyzer = HTTPTapAnalyzer()
steps = analyzer.analyze_url("https://httpbin.io")
YAMLExporter().export(steps, "https://httpbin.io", "output.yaml")

RequestExecutor¶

For full control over how requests are performed, implement RequestExecutor and pass an instance as request_executor= to HTTPTapAnalyzer.

RequestExecutor ¶

Bases: Protocol

Protocol describing modern request executors used by the analyzer.

Implementations perform a single HTTP request described by a :class:RequestOptions instance and return the collected metrics as a :class:RequestOutcome. This lets the analyzer delegate the actual transport work to interchangeable backends.

示例：

>>> class CustomExecutor:
...     def execute(self, options: RequestOptions) -> RequestOutcome:
...         ...  # perform the request and collect metrics

方法：

execute –

Perform an HTTP request based on the provided options.

execute ¶

execute(options: RequestOptions) -> RequestOutcome

Perform an HTTP request based on the provided options.

参数：

options (RequestOptions) –

Fully populated request parameters, including URL, timeout, method, and any injected collaborators.

返回：

RequestOutcome –

A RequestOutcome bundling the timing, network, and response data.

引发：

HTTPClientError –

If the request cannot be completed. Implementations should surface transport failures using this error type so the analyzer can record partial data.

源代码位于： httptap/request_executor.py

def execute(self, options: RequestOptions) -> RequestOutcome:
    """Perform an HTTP request based on the provided options.

    Args:
        options: Fully populated request parameters, including URL,
            timeout, method, and any injected collaborators.

    Returns:
        A RequestOutcome bundling the timing, network, and response data.

    Raises:
        httptap.http_client.HTTPClientError: If the request cannot be
            completed. Implementations should surface transport failures
            using this error type so the analyzer can record partial data.
    """

RequestOptions `dataclass` ¶

RequestOptions(
    url: str,
    timeout: float,
    method: HTTPMethod = GET,
    content: bytes | None = None,
    http2: bool = True,
    verify_ssl: bool = True,
    ca_bundle_path: str | None = None,
    dns_resolver: DNSResolver | None = None,
    tls_inspector: TLSInspector | None = None,
    timing_collector: TimingCollector | None = None,
    force_new_connection: bool | None = None,
    headers: Mapping[str, str] | None = None,
    proxy: ProxyTypes | None = None,
    noproxy: bool = False,
    deadline: float | None = None,
)

Aggregates all parameters required to perform a single HTTP request.

属性：

url (str) –

Target URL to request. Must be a valid HTTP/HTTPS URL.
timeout (float) –

Request timeout in seconds.
deadline (float | None) –

Optional monotonic deadline shared by a request chain.
method (HTTPMethod) –

HTTP method to use for the request.
content (bytes | None) –

Optional request body as bytes.
http2 (bool) –

Whether to enable HTTP/2 support.
verify_ssl (bool) –

Whether to verify TLS certificates.
ca_bundle_path (str | None) –

Path to a custom CA certificate bundle (PEM format). Only used when verify_ssl is True. If None, the system CA bundle is used.
dns_resolver (DNSResolver | None) –

Custom DNS resolver implementation. If None, the executor uses its default resolver.
tls_inspector (TLSInspector | None) –

Custom TLS inspector implementation, used only for the fallback probe when the live connection exposes no TLS data and no proxy is in use. If None, the executor uses its default inspector.
timing_collector (TimingCollector | None) –

Timing collector instance used to measure request phases. If None, a fresh PerfCounterTimingCollector is used.
force_new_connection (bool | None) –

Deprecated and ignored. A new connection is always used because each request gets a fresh client. Accepted only for backward compatibility; passing a value emits a DeprecationWarning.
headers (Mapping[str, str] | None) –

Optional mapping of request headers to send.
proxy (ProxyTypes | None) –

Optional proxy URL (http/https/socks5/socks5h) applied to the request.
noproxy (bool) –

When True, ignore proxy environment variables and connect directly.

RequestOutcome `dataclass` ¶

RequestOutcome(
    timing: TimingMetrics,
    network: NetworkInfo,
    response: ResponseInfo,
)

Wraps the collected timing, network, and response objects.

属性：

timing (TimingMetrics) –

Timing metrics gathered for the request phases.
network (NetworkInfo) –

Network and TLS/certificate information for the connection.
response (ResponseInfo) –

HTTP response metadata (status, headers, body size).

Type checking¶

All protocols are fully type-hinted and work with mypy, pyright, and other type checkers. Because they are structural, any class implementing the required methods satisfies the type — no explicit subclassing needed.

from httptap.interfaces import DNSResolver


class MyResolver:
    def resolve(self, host: str, port: int, timeout: float) -> tuple[str, str, float]:
        return "192.168.1.1", "IPv4", 10.5


resolver: DNSResolver = MyResolver()  # verified by the type checker

Next steps¶

See core components documentation
Review advanced usage examples
Check contributing guidelines to add new protocols

Protocol Interfaces¶

Why protocols?¶

DNSResolver¶

DNSResolver ¶

resolve ¶

Example implementation¶

Address fallback with resolve_all()¶

TLSInspector¶

TLSInspector ¶

inspect ¶

Example implementation¶

TimingCollector¶

TimingCollector ¶

mark_dns_start ¶

mark_dns_end ¶

mark_request_start ¶

mark_ttfb ¶

mark_request_end ¶

get_metrics ¶

Example implementation¶

Visualizer¶

Visualizer ¶

render ¶

Example implementation¶

Exporter¶

Exporter ¶

export ¶

Example implementation¶

RequestExecutor¶

RequestExecutor ¶

execute ¶

RequestOptions dataclass ¶

RequestOutcome dataclass ¶

Type checking¶

Next steps¶

Address fallback with `resolve_all()`¶

RequestOptions `dataclass` ¶

RequestOutcome `dataclass` ¶