Flowsint Enricher Builder

Flowsint Enricher Builder

You build enrichers and types for Flowsint. You do not memorize the catalog — you know where to look and how the pieces fit. Always read source before generating code: type definitions and existing enrichers are the ground truth.

Authoritative source paths

Read these first. Never assume signatures or fields — open the file.

The first question: new type or reuse?

When the user describes a enricher, decide before writing code:

1. List the entities involved — input data, output data, intermediate fields you'll attach.
2. For each, check flowsint-types/src/flowsint_types/ — open the closest candidate file and read its fields.
3. Decide:
   • Reuse if existing type covers all required fields (extras allowed — ConfigDict.extra = "allow").
   • Extend an existing type if 1–2 fields are missing — propose adding optional fields to the existing model.
   • Create new type if the entity is conceptually distinct (different primary key, different label semantics, different graph role).
4. Never cram data into a wrong type. If a "Domain" enricher returns risk scores, a RiskProfile exists — don't stuff scores into Domain metadata. If nothing fits, propose a new type and tell the user why.

Surface the decision to the user before generating code: list candidate types you found, what's missing, and your recommendation.

Anatomy of an enricher

Minimum surface (read enricher_base.py for the full contract):

from typing import List
from flowsint_core.core.enricher_base import Enricher
from flowsint_core.core.logger import Logger
from flowsint_enrichers.registry import flowsint_enricher
from flowsint_types import Domain, Ip  # or whatever types


@flowsint_enricher
class MyEnricher(Enricher):
    """[Source name] One-line purpose."""

    InputType = Domain  # base type, not List[Domain]
    OutputType = Ip

    @classmethod
    def name(cls) -> str:
        return "domain_to_ip"  # snake_case, unique

    @classmethod
    def category(cls) -> str:
        return "Domain"  # see note on casing below

    @classmethod
    def key(cls) -> str:
        return "domain"  # primary field of InputType

    @classmethod
    def get_params_schema(cls):  # optional, only if params needed
        return [...]

    async def scan(self, data: List[InputType]) -> List[OutputType]: ...

    def postprocess(self, results, input_data):
        for src, dst in zip(input_data, results):
            self.create_node(src)
            self.create_node(dst)
            self.create_relationship(src, dst, "RESOLVES_TO")
        return results


InputType = MyEnricher.InputType
OutputType = MyEnricher.OutputType

File location: flowsint-enrichers/src/flowsint_enrichers/<input_type>/to_<target>.py. The directory matches the input type's lowercase name. If no directory exists for your input type, create it (no __init__.py needed — auto-discovery walks the tree).

Registration: the @flowsint_enricher decorator does it. Do not edit any registry.py. API restart picks up new files via load_all_enrichers().

Params and secrets

Defined via get_params_schema() classmethod. Each entry is a dict:

Read params inside scan():

mode = self.params.get("mode", "passive")
api_key = self.get_secret("MY_API_KEY")  # vault-resolved during async_init

Vault resolution flow (see Enricher.resolve_params in enricher_base.py):

1. If user passed a vault ID in params → vault looked up by that ID.
2. Else → vault looked up by the param name (e.g. MY_API_KEY).
3. If required: true and nothing found → Exception("Required vault secret 'MY_API_KEY' is missing...").

Never hardcode keys. Always declare a vaultSecret param. Document the expected vault key name in the docstring.

Graph operations (postprocess)

create_node(obj) and create_relationship(from_obj, to_obj, rel_label="IS_RELATED_TO") take Pydantic objects directly. Don't manually construct node dicts — pass the typed instance.

Relationship label convention: UPPER_SNAKE_CASE verb phrase (HAS_DOMAIN, RESOLVES_TO, FOUND_IN_BREACH). Be consistent with existing enrichers — grep before inventing a new label.

self.log_graph_message("...") for graph-related progress logs. Logger.info / error / warn(self.sketch_id, {"message": "..."}) for general logs.

Creating a new type — checklist

When you decide a new type is warranted:

1. File: flowsint-types/src/flowsint_types/<snake_case>.py.
2. Class: PascalCase, inherit from FlowsintType, decorate with @flowsint_type.
3. Exactly one primary field: Field(..., json_schema_extra={"primary": True}). Must uniquely identify the entity (used as Neo4j MERGE key).
4. compute_label: @model_validator(mode='after'), sets self.nodeLabel, returns self. Handle None for optional fields.
5. Export in __init__.py: add import + entry in __all__.
6. Category (optional but recommended): add a ("MyType", "primary_field_name", icon) tuple in _get_category_definitions() in type_registry_service.py. Without this, the type works as an enricher I/O but doesn't show in the UI type picker.
7. Reinstall: cd flowsint-types && poetry install (or make prod from repo root).
8. Test: write a tests/test_<name>.py covering creation, primary uniqueness, compute_label with full/partial fields.

Full template + patterns: docs/developers/managing-types.mdx.