# How to Integrate External Tools Like DNSx as Flowsint Enrichers > Learn how to integrate external tools like DNSx as Flowsint enrichers. Create DockerTool and Enricher subclasses to extend Flowsint capabilities and automate your data enrichment. - Repository: [reconurge/flowsint](https://github.com/reconurge/flowsint) - Tags: how-to-guide - Published: 2026-06-05 --- **To integrate external tools like DNSx as Flowsint enrichers, create a `DockerTool` subclass that wraps the DNSx container, build an `Enricher` subclass that invokes the tool and maps JSON output to Flowsint core types, and place both files under the `flowsint-enrichers/src/tools/` and `flowsint-enrichers/src/enrichers/` directories so the registry discovers them automatically.** Flowsint, maintained in the `reconurge/flowsint` repository, separates low-level command execution from high-level graph enrichment to keep reconnaissance pipelines modular. Understanding how to integrate external tools as Flowsint enrichers lets you bring any Dockerized utility—such as the ProjectDiscovery DNSx resolver—into the platform as a first-class component. ## Understand the Tool and Enricher Separation The architecture is split into two distinct layers. The **tool layer** handles the actual execution of an external program, while the **enricher layer** orchestrates that execution and translates raw results into persisted graph nodes. - **`Tool` base class** — Defined in [`flowsint-enrichers/src/tools/base.py`](https://github.com/reconurge/flowsint/blob/main/flowsint-enrichers/src/tools/base.py), this abstract class establishes the metadata contract and `launch` signature every tool must implement. - **`DockerTool` helper** — Located in [`flowsint-enrichers/src/tools/dockertool.py`](https://github.com/reconurge/flowsint/blob/main/flowsint-enrichers/src/tools/dockertool.py), it provides reusable Docker lifecycle methods including `is_installed`, `install`, and a generic `launch` that runs the container and captures `stdout`. - **`Enricher` base class** — Imports the tool, iterates over source entities (such as `Domain`), and converts output dictionaries into strongly typed models from `flowsint_core.types`. - **Registry** — The core registry at [`flowsint-core/src/flowsint_core/core/services/registry.py`](https://github.com/reconurge/flowsint/blob/main/flowsint-core/src/flowsint_core/core/services/registry.py) auto-discovers enrichers placed under `flowsint-enrichers/src/enrichers/`, making them available to flow definitions without manual wiring. ## Step 1: Create a DockerTool Subclass for DNSx The first concrete task is a tool wrapper that knows how to pull the `projectdiscovery/dnsx` image, build the command line, and parse newline-delimited JSON. Save this implementation as [`flowsint-enrichers/src/tools/network/dnsx.py`](https://github.com/reconurge/flowsint/blob/main/flowsint-enrichers/src/tools/network/dnsx.py). ```python # src/tools/network/dnsx.py from tools.dockertool import DockerTool from typing import List, Optional, Any import json class DnsxTool(DockerTool): """Wrapper for the DNSx Docker image (projectdiscovery/dnsx).""" image = "projectdiscovery/dnsx" default_tag = "latest" @classmethod def name(cls) -> str: return "dnsx" @classmethod def category(cls) -> str: return "Network" @classmethod def description(cls) -> str: return "Fast DNS resolver/subdomain enumeration (ProjectDiscovery DNSx)." @classmethod def version(cls) -> str: return "2.2.0" def launch( self, target: str, json_output: bool = True, resolver: Optional[str] = None, timeout: int = 30, ) -> List[dict]: """ Execute DNSx against ``target`` and return parsed JSON records. Args: target: Domain/IP or a file containing a list of targets. json_output: Ask DNSx to emit newline‑delimited JSON. resolver: Custom DNS resolver (e.g. ``8.8.8.8:53``). timeout: Maximum container runtime in seconds. Returns: List of dictionaries, each representing a DNS record. """ # Ensure the image is available if not self.is_installed(): self.install() # Build the command line cmd = f"-d {target}" if json_output: cmd += " -json" if resolver: cmd += f" -r {resolver}" # Run the container raw = super().launch(command=cmd, timeout=timeout) # Parse newline‑delimited JSON results: List[dict] = [] for line in raw.strip().split("\n"): if not line: continue try: results.append(json.loads(line)) except json.JSONDecodeError: # Skip malformed lines – tool is defensive continue return results ``` **Key implementation details:** - **`DockerTool` lifecycle** — The wrapper inherits container pull, run, and cleanup behavior, so `is_installed()` and `install()` work without extra code. - **Command builder pattern** — The `cmd` string is assembled conditionally based on method arguments, mirroring the pattern documented in `docs/developers/managing-tools.mdx`. - **Output parser pattern** — Raw `stdout` is split into lines and each line is parsed individually, making the tool resilient to partial or malformed output. ## Step 2: Build the Enricher to Map DNSx Output to Graph Entities Next, create an enricher that consumes the tool and converts DNS records into Flowsint core types. Save this file as [`flowsint-enrichers/src/enrichers/network/dnsx_enricher.py`](https://github.com/reconurge/flowsint/blob/main/flowsint-enrichers/src/enrichers/network/dnsx_enricher.py). ```python # src/enrichers/network/dnsx_enricher.py from enrichers.base import Enricher from tools.network.dnsx import DnsxTool from flowsint_core.types import Domain, Ip, Cname, DnsRecord # Pydantic models from typing import List class DnsxEnricher(Enricher): """Enricher that resolves a domain (or a list) via DNSx.""" async def scan(self, domains: List[Domain]) -> List[DnsRecord]: tool = DnsxTool() records: List[DnsRecord] = [] for domain in domains: raw = tool.launch(target=domain.name, json_output=True) for entry in raw: # DNSx returns a dict with keys like "type", "name", "value" rec_type = entry.get("type") name = entry.get("name") value = entry.get("value") if rec_type == "A": records.append(Ip(address=value, source=domain)) elif rec_type == "CNAME": records.append(Cname(name=name, target=value, source=domain)) else: # Generic record type – store as DnsRecord for later enrichment records.append( DnsRecord( type=rec_type, name=name, value=value, source=domain, ) ) return records ``` **What happens during enrichment:** 1. The `scan` method receives a list of `Domain` objects. 2. For each domain, it calls `DnsxTool.launch` with `json_output=True`. 3. Returned JSON rows are mapped onto **Flowsint core types** (`Ip`, `Cname`, `DnsRecord`) according to the DNS record type. 4. The framework automatically persists those objects and creates graph edges—for example, `DOMAIN → RESOLVES_TO → IP`—in Neo4j via the core service API. ## Step 3: Register the Enricher and Reference It in a Flow Flowsint uses entry-point style discovery for enrichers. If you place [`dnsx_enricher.py`](https://github.com/reconurge/flowsint/blob/main/dnsx_enricher.py) under `flowsint-enrichers/src/enrichers/network/`, the registry picks it up automatically at startup. For explicit control, you can add it to the `ENRICHERS` mapping in [`flowsint-enrichers/src/enrichers/__init__.py`](https://github.com/reconurge/flowsint/blob/main/flowsint-enrichers/src/enrichers/__init__.py): ```python from .network.dnsx_enricher import DnsxEnricher ENRICHERS = { "dnsx": DnsxEnricher, # …other enrichers… } ``` Once registered, users can reference the enricher by name inside a flow definition: ```yaml steps: - name: dnsx type: enricher inputs: - domain: mytarget.com ``` The engine resolves `"dnsx"` through the registry, executes the `DnsxEnricher.scan` coroutine, and merges the results into the active investigation graph. ## Summary - **Tool layer** — Subclass `DockerTool` (from [`flowsint-enrichers/src/tools/dockertool.py`](https://github.com/reconurge/flowsint/blob/main/flowsint-enrichers/src/tools/dockertool.py)) to wrap external Docker images, implement `launch`, and parse raw output. - **Enricher layer** — Subclass `Enricher` (from `enrichers.base`), invoke your tool per source entity, and map JSON records to Flowsint types such as `Ip`, `Cname`, and `DnsRecord`. - **Registration** — Drop files under `flowsint-enrichers/src/enrichers/` for auto-discovery, or map them explicitly in the package [`__init__.py`](https://github.com/reconurge/flowsint/blob/main/__init__.py). - **Flow usage** — Reference the enricher by its registered name in YAML flow definitions; the engine handles scheduling and Neo4j persistence automatically. ## Frequently Asked Questions ### What is the difference between a Tool and an Enricher in Flowsint? A **Tool** is a low-level wrapper that knows how to execute an external binary or container and return raw data. An **Enricher** is a high-level orchestrator that accepts source graph entities, calls one or more tools, validates the output, and converts it into typed nodes that Flowsint persists in Neo4j. ### Where does the registry discover new enrichers? The registry defined in [`flowsint-core/src/flowsint_core/core/services/registry.py`](https://github.com/reconurge/flowsint/blob/main/flowsint-core/src/flowsint_core/core/services/registry.py) walks the `flowsint-enrichers/src/enrichers/` package at startup. Any `Enricher` subclass placed inside that tree is loaded automatically via entry-point discovery. ### How does Flowsint persist enricher output? When an enricher returns typed objects—such as `Ip` or `Cname`—the core service API automatically writes them to Neo4j and creates appropriate graph edges. You do not need to write Cypher queries inside the enricher; the framework handles persistence once the typed models are returned from `scan`. ### Can the DockerTool pattern be reused for other command-line utilities? Yes. The `DockerTool` base class in [`flowsint-enrichers/src/tools/dockertool.py`](https://github.com/reconurge/flowsint/blob/main/flowsint-enrichers/src/tools/dockertool.py) is generic. You can subclass it for any utility distributed as a Docker image—such as other ProjectDiscovery tools—by setting the `image` and `default_tag` class attributes and implementing a `launch` method that builds the tool-specific command line and parses its output.