How to Set Up Email Enrichment Pipelines in Flowsint: A Complete Guide

To set up email enrichment pipelines in Flowsint, instantiate an Email entity, store required API keys in the vault, define a YAML template that lists the desired enrichers, and execute the template via TemplateEnricher.run() or the CLI.

Configuring email enrichment pipelines in the reconurge/flowsint repository lets you transform raw email addresses into structured OSINT graph nodes. The framework provides purpose-built email enrichers—such as EmailToDomainsEnricher and EmailToLeaksEnricher—that are orchestrated by the core TemplateEnricher service. In this guide, you will learn how to wire these components together using secrets management, YAML templates, and the Neo4j-backed graph store.

How Email Enrichment Works in Flowsint

Flowsint’s enrichment architecture revolves around the TemplateEnricher class located in flowsint_core/core/template_enricher.py. This orchestrator parses a YAML pipeline definition and dispatches work to specialized email enrichers registered in flowsint_enrichers/registry.py.

Each enricher implements the BaseEnricher interface and performs a single task—resolving domains, checking breaches, generating Gravatar URLs, or extracting usernames. The enrichers retrieve external API credentials through the VaultService (flowsint_core/core/vault.py) and persist results as Neo4j nodes and relationships via create_node and create_relationship helpers.

Step-by-Step: Configure an Email Enrichment Pipeline

Step 1 — Model the Input with the Email Entity

Every pipeline starts with a validated Email entity from flowsint-types/src/flowsint_types/email.py. The Pydantic model uses EmailStr for validation and sets its nodeLabel property to the raw email address. This value becomes the primary identifier for the resulting Neo4j node.

Step 2 — Register API Keys in the Vault

Flowsint never hard-codes third-party credentials. Instead, the VaultService reads secrets from the process environment or an encrypted store. For local development, place keys in a .env file:

WHOXY_API_KEY=your_whoxy_key
HIBP_API_KEY=your_hibp_key

At runtime, the VaultService in flowsint_core/core/vault.py injects these values into the enrichers that request them. You do not need to manually pass keys to individual enricher constructors.

Step 3 — Declare the YAML Enrichment Template

The TemplateEnricher expects a YAML file that enumerates which enrichers to run and in what order. A standard email pipeline template looks like this:

enrich:
  - EmailToDomainsEnricher
  - EmailToLeaksEnricher
  - EmailToGravatarEnricher
  - EmailToUsernameEnricher

The TypeRegistryService resolves each string name to its concrete enricher class before the pipeline begins execution. This decoupling lets you compose pipelines without importing classes directly.

Run Email Enrichment from Code or CLI

You can launch the pipeline from Python or the command line. The following examples show the three most common execution patterns.

Launch a Full Pipeline in Python

Begin by instantiating the Email model. Then load the YAML template into TemplateEnricher and call run() to start the pipeline:

from flowsint_core.core.template_enricher import TemplateEnricher
from flowsint_types.email import Email

email = Email(email="jane.doe@example.com")

enricher = TemplateEnricher(template_path="templates/email_pipeline.yaml")
await enricher.run(initial_entity=email)

The orchestrator resolves each enricher through TypeRegistryService, retrieves secrets from VaultService, and writes enriched nodes to Neo4j. All of this happens asynchronously, so you should await the call.

Debug a Single Enricher Manually

For targeted testing, import an enricher directly and run it against a single entity. This bypasses the template layer and is useful when iterating on a specific data source:

from flowsint_enrichers.email.to_domains import EmailToDomainsEnricher
from flowsint_core.core.services.vault_service import VaultService

VaultService().set_secret("WHOXY_API_KEY", "my_key")

email = Email(email="john@example.org")
enricher = EmailToDomainsEnricher()
await enricher.enrich(email)

The example above sets the WHOXY_API_KEY explicitly via VaultService().set_secret(). In production, the vault loads the secret automatically from the environment.

Execute from the Command Line

Flowsint ships with a lightweight CLI wrapper that builds the entity and hands it to TemplateEnricher automatically. Pass the template path and a JSON entity payload as shown below:

flowsint run --template templates/email_pipeline.yaml \
             --entity '{"type":"Email","email":"alice@example.com"}'

The CLI parses the JSON payload into the correct Pydantic model. It then starts the pipeline without requiring any additional Python boilerplate.

Available Email Enrichers in flowsint-enrichers

The email package inside flowsint-enrichers/src/flowsint_enrichers/email/ contains four production-ready enrichers. Each one registers itself automatically with the global EnricherRegistry so that TemplateEnricher can discover it at runtime.

• EmailToDomainsEnricher (to_domains.py) — Queries the WHOXY API using WHOXY_API_KEY and creates HAS_DOMAIN relationships linking the email to discovered domains.

• EmailToLeaksEnricher (to_leaks.py) — Contacts the Have-I-Been-Pwned API with HIBP_API_KEY and attaches BREACHED_IN edges to Leak nodes.

• EmailToGravatarEnricher (to_gravatar.py) — Computes the MD5 hash of the address to build a Gravatar profile URL. No external API key is required.

• EmailToUsernameEnricher (to_username.py) — Parses the local-part of the email address to generate a Username node connected by a HAS_USERNAME relationship.

Summary

• Email enrichment pipelines in Flowsint are driven by the TemplateEnricher class, which reads a YAML list of enrichers and runs them sequentially.
• The Email model in flowsint-types/src/flowsint_types/email.py validates addresses and supplies a nodeLabel for Neo4j.
• Secrets such as WHOXY_API_KEY and HIBP_API_KEY are injected via the VaultService—typically from a .env file in development.
• Four built-in email enrichers—EmailToDomainsEnricher, EmailToLeaksEnricher, EmailToGravatarEnricher, and EmailToUsernameEnricher—handle domain resolution, breach checking, avatar generation, and username extraction.
• You can trigger pipelines programmatically with TemplateEnricher.run(), manually via individual enricher classes, or through the flowsint run CLI.

Frequently Asked Questions

What API keys are required to set up email enrichment pipelines in Flowsint?

You need WHOXY_API_KEY for EmailToDomainsEnricher and HIBP_API_KEY for EmailToLeaksEnricher. The EmailToGravatarEnricher and EmailToUsernameEnricher do not require external API keys.

How does Flowsint store secrets for email enrichment?

Flowsint uses the VaultService located in flowsint_core/core/vault.py to manage credentials. In development, you can store keys in a .env file, and the vault injects them into the process environment at runtime.

Can I run a single email enricher without using the TemplateEnricher?

Yes. Each enricher exposes a direct enrich() method. You can instantiate EmailToDomainsEnricher manually, pass it a single Email entity, and call await enricher.enrich(email) without involving TemplateEnricher.

Where does Flowsint save the output of an email enrichment pipeline?

All enrichment results are persisted as Neo4j graph nodes and relationships. The base enricher class provides create_node and create_relationship helpers that write entities such as Domain, Leak, or Username and link them back to the original Email node.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}