hydrogen/HYDROGEN_TRANSPORTS.md

Hydrogen Custom Report Transports

This document explains how report transports work in Hydrogen, how the built-in transports behave, and what you need to do to add your own transport implementation.

Where Transports Fit in the Hydrogen Pipeline

Hydrogen builds a report after all security modules finish.

The reporting flow in reporting/service.py is:

1. Build an AuditReport.
2. Render it into a RenderedReport.
3. Publish the rendered report through a transport.

The relevant method is:

def publish(self, report: AuditReport, report_config: ResolvedReportOutput) -> str:
    rendered_report = self.render(report, report_config)
    transport = self._transport_registry.get(report_config.transport.type)
    return transport.publish(rendered_report, report_config.transport)

This means transports operate on already-rendered content. They do not receive raw findings directly.

Core Transport Contract

The base transport contract lives in core/base.py.

class BaseTransport(ABC):
    transport_type: str
    config_model: type[BaseModel] = EmptyPluginConfig

    @abstractmethod
    def publish(self, rendered_report: RenderedReport, transport: ResolvedPluginConfig) -> str: ...

Every transport must define:

• transport_type — the name used in config.yaml.
• config_model — optional Pydantic model for transport-specific configuration.
• publish(...) — the delivery implementation.

The transport parameter is a ResolvedPluginConfig:

@dataclass(frozen=True)
class ResolvedPluginConfig:
    type: str
    config: BaseModel

• transport.type — the transport type string.
• transport.config — the validated config model instance.

The return value of publish() is a string. The built-in transports return values such as the output path or an HTTP status string. Hydrogen currently does not use that return value elsewhere, but your transport still needs to return one.

Typed Transport Pattern

Hydrogen provides TypedTransport as a convenient base class for transports with their own config model:

class TypedTransport(BaseTransport, Generic[TReportTransport], ABC):
    config_model: type[TReportTransport]

    def publish(self, rendered_report: RenderedReport, transport: ResolvedPluginConfig) -> str:
        if not isinstance(transport.config, self.config_model):
            raise TypeError(
                f"Transport '{self.transport_type}' expected {self.config_model.__name__}, "
                f"got {type(transport.config).__name__}"
            )
        return self.publish_typed(rendered_report, cast(TReportTransport, transport.config))

    @abstractmethod
    def publish_typed(self, rendered_report: RenderedReport, transport: TReportTransport) -> str: ...

This is the recommended pattern when your transport has its own configuration model.

Benefits:

• Runtime type checking for the config object.
• Cleaner transport code — publish_typed() receives the typed config directly.
• No manual casting inside publish().

How Hydrogen Discovers Transports

Transport discovery is implemented in reporting/bootstrap.py and core/plugin_types.py.

Hydrogen discovers transports from two sources:

1. Package scan — Python packages listed in plugin_packages.transports (default: ["reporting.transports"]).
2. Entry points — hydrogen.transports entry points registered by installed packages.

transport_descriptors, transport_errors = discover_plugins(
    PluginGroupConfig(
        package_names=plugin_packages.transports,
        entry_point_group="hydrogen.transports",
        plugin_kind="transport",
    ),
    BaseTransport,
)

The discovery process:

• Imports the package and all its submodules via pkgutil.walk_packages.
• Inspects each module for non-abstract subclasses of BaseTransport.
• Filters out classes that are only imported (not defined) in that module.
• Instantiates each discovered class with no constructor arguments.
• Registers them in TransportRegistry by transport_type.

Important consequences for custom transports:

• Your class must be a non-abstract subclass of BaseTransport.
• The class must be defined in the module itself, not only imported from another module.
• The class constructor must work with no arguments.
• Registration is based on transport_type — names must be unique.

Built-In Transport: file

Implementation: reporting/transports/file.py

Config model: FileTransportConfig

class FileTransportConfig(BaseModel):
    path: str = Field()
    append_extension: bool = Field(True)

Configuration:

reports:
  outputs:
    - renderer:
        type: json
      transport:
        type: file
        path: reports/latest
        append_extension: true

Behavior:

• path is converted to Path.
• If append_extension is true and the destination has no suffix, Hydrogen appends the renderer's file extension.
• Parent directories are created automatically.
• The report is written as UTF-8 text.

destination = Path(transport.path)
if transport.append_extension and not destination.suffix:
    destination = destination.with_suffix(rendered_report.file_extension)

destination.parent.mkdir(parents=True, exist_ok=True)
destination.write_text(rendered_report.content, encoding="utf-8")

Practical example:

• path: reports/audit with the JSON renderer becomes reports/audit.json.
• path: reports/audit.txt stays reports/audit.txt.

Built-In Transport: webhook

Implementation: reporting/transports/webhook.py

Config model: WebhookTransportConfig

class WebhookTransportConfig(BaseModel):
    url: AnyHttpUrl = Field()
    method: WebhookMethod = Field(WebhookMethod.POST)
    headers: dict[str, str] = Field(default_factory=dict)
    timeout_seconds: float = Field(10.0, gt=0)
    payload_mode: WebhookPayloadMode = Field(WebhookPayloadMode.RENDERED)

Configuration:

reports:
  outputs:
    - renderer:
        type: json
      transport:
        type: webhook
        url: https://example.internal/ingest
        method: POST
        headers:
          Authorization: Bearer token
        timeout_seconds: 10
        payload_mode: rendered

Supported fields:

• url — validated as HTTP or HTTPS URL by Pydantic.
• method — POST, PUT, or PATCH.
• headers — arbitrary string-to-string HTTP headers.
• timeout_seconds — positive float.
• payload_mode — rendered or envelope.

payload_mode: rendered

Hydrogen sends the renderer output as the raw HTTP body.

Header behavior:

• Existing custom headers are preserved.
• Content-Type defaults to the renderer media type if you did not set it explicitly.

For the built-in JSON renderer, that means Content-Type: application/json unless overridden.

payload_mode: envelope

Hydrogen sends a JSON object with report metadata and content:

{
  "format": "json",
  "content_type": "application/json",
  "file_extension": ".json",
  "content": "...rendered report..."
}

Header behavior:

• Content-Type defaults to application/json.

Adding a Custom Transport

To add a custom transport, you need two things:

1. A transport config model.
2. A transport class implementing BaseTransport or TypedTransport.

Via Package (In-Repository)

Create a file under reporting/transports/:

reporting/transports/stdout.py:

from pydantic import BaseModel, Field

from core.base import TypedTransport
from reporting.models import RenderedReport


class StdoutTransportConfig(BaseModel):
    prefix: str = Field("[Hydrogen]")


class StdoutTransport(TypedTransport[StdoutTransportConfig]):
    transport_type = "stdout"
    config_model = StdoutTransportConfig

    def publish_typed(self, rendered_report: RenderedReport, transport: StdoutTransportConfig) -> str:
        print(f"{transport.prefix}\n{rendered_report.content}")
        return "stdout"

Use it in config.yaml:

reports:
  outputs:
    - renderer:
        type: json
      transport:
        type: stdout
        prefix: "[Hydrogen Report]"

Via Entry Points (Third-Party Package)

If you distribute your transport as a pip-installable package, register it via entry points:

[project.entry-points."hydrogen.transports"]
stdout = "my_package.transports:StdoutTransport"

The entry point value can be:

• A class reference (Hydrogen instantiates it).
• A pre-built instance.

Hydrogen validates that the result is a non-abstract subclass or instance of BaseTransport.

Via plugin_packages Configuration

You can add additional packages for transport discovery:

plugin_packages:
  transports:
    - reporting.transports
    - mycompany.transport_extras

Hydrogen scans all listed packages and their submodules for BaseTransport subclasses.

Configuration Schema for Custom Transports

Unlike the old architecture (which required a Pydantic discriminated union), transports no longer need to be added to a central type union. The new PluginConfigRef model uses extra="allow":

class PluginConfigRef(BaseModel):
    model_config = ConfigDict(extra="allow")
    type: str = Field(min_length=1)

    def payload(self) -> dict[str, Any]:
        return dict(self.model_extra or {})

This means:

• The type field selects the transport by name.
• Any additional YAML fields are captured as the "payload".
• Hydrogen validates the payload against your transport's config_model at runtime.

There is no central transport union to modify. As long as your transport is discovered and registered, and its config_model matches the provided payload, it will work.

The __init__.py Export

You may export your custom transport from reporting/transports/__init__.py, but Hydrogen discovery does not depend on that export.

Why:

• Hydrogen walks the package tree and imports submodules directly.
• Discovery inspects those imported submodules, not just the __init__.py exports.

So this file is optional for plugin loading:

from reporting.transports.stdout import StdoutTransport

It is still useful if you want a cleaner package API for developers.

Transport Name Collisions

TransportRegistry.register() stores transports in a dictionary by transport_type:

self._transports[transport.transport_type] = transport

This means:

• Names must be unique.
• If two transports declare the same transport_type, register() raises a ValueError.

In bootstrap.py, collisions are caught and converted to PluginLoadError:

try:
    transport_registry.register(descriptor.plugin)
except ValueError as exc:
    errors.append(PluginLoadError(plugin_kind="transport", source=descriptor.source, message=str(exc)))

Choose stable, explicit names for custom Hydrogen transports.

Limits of the Current Hydrogen Transport Design

Before designing a transport extension, be aware of the current limits:

• Transports are instantiated with no constructor arguments.
• There is no dependency injection container.
• There is no built-in retry, queueing, or backoff abstraction.
• The transport type string must match transport_type exactly.

These limits do not make custom transports impossible. They just mean the extension point is code-first rather than fully pluggable from YAML alone.

Troubleshooting Custom Transports

If a transport does not work in Hydrogen, check these points first:

1. The class subclasses BaseTransport or TypedTransport.
2. The class is defined inside a module under one of the configured transport packages.
3. The class is not abstract.
4. The class can be instantiated with no arguments.
5. transport_type matches the YAML transport.type value.
6. The transport's config_model validates the YAML payload correctly.
7. The renderer selected by renderer.type exists and runs successfully.
8. If using extra YAML fields, they are validated by the transport's config_model (not by PluginConfigRef).

Summary

In Hydrogen, a transport is the final delivery mechanism for a rendered report. The implementation is straightforward:

• A transport class discoverable from configured packages or entry points.
• An optional config model validated at runtime via PluginConfigRef.

If you remember that discovery and configuration are separate concerns, extending Hydrogen transports becomes predictable and low-risk. Unlike the old architecture, there is no central transport union to modify — any discovered transport with a valid config_model is immediately usable.