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.

示例：

>>> 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_'.

引发：

Exception –

If hostname cannot be resolved. Implementations should define specific exception types for DNS failures.

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:
        Exception: If hostname cannot be resolved. Implementations should
            define specific exception types for DNS failures.

    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.

Example implementation¶

import socket
import time


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)
            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
        except socket.gaierror as e:
            raise Exception(f"DNS resolution failed: {e}")


from httptap import HTTPTapAnalyzer

analyzer = HTTPTapAnalyzer(dns_resolver=CustomDNSResolver())

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).

    """

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 = True,
    headers: Mapping[str, str] | None = None,
    proxy: ProxyTypes | None = None,
    noproxy: bool = False,
)

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.
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. If None, the executor uses its default inspector.
timing_collector (TimingCollector | None) –

Timing collector instance used to measure request phases. If None, no phase timing is collected for this request.
force_new_connection (bool) –

Whether to force a fresh connection instead of reusing a pooled one, ensuring per-request timing is accurate.
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¶

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¶

RequestOptions `dataclass` ¶

RequestOutcome `dataclass` ¶