Ingesting Multi-Channel Notification Workflows into Cognee Knowledge Graphs with Novu

Modern applications communicate with users across multiple channels, including email, SMS, push notifications, in-app messages, and chat platforms.

As these systems grow, notification infrastructure becomes more complex. Workflows trigger messages, topics organize subscribers, and provider integrations handle delivery through external services.

Novu provides notification infrastructure for building and managing these workflows. But as an application accumulates more notification rules and integrations, answering architectural questions becomes harder.

Engineering, product, and support teams might need to know:

• Which workflows send critical alerts through SMS or chat?
• Which delivery providers are configured and active?
• Which topics exist, and how many subscribers are associated with them?
• Which workflows depend on a particular provider integration?

Answering these questions often requires navigating multiple dashboards and API resources.

What if an AI agent could retrieve this notification architecture from a shared knowledge graph?

That's the idea behind cognee-community-connector-novu, a community connector designed to synchronize Novu notification metadata into Cognee using dlt.

In this article, we'll explore the connector's architecture, ingestion workflow, entity representation, security considerations, and testing strategy.

1. Architecture: Bringing Notification Infrastructure into AI Memory

The connector resides in the Cognee Community repository under:

packages/connector/novu

Its intended architecture connects three components:

• Novu: Provides notification workflow, topic, group, and integration metadata.
• dlt: Handles structured data ingestion and synchronization.
• Cognee: Processes ingested information for knowledge-graph and semantic retrieval.

The data flow looks like this:

Novu API
    |
    v
Novu Data-Source Connector
    |
    v
dlt Ingestion Pipeline
    |
    v
Cognee Memory Processing
    |
    v
Knowledge Graph and Search
    |
    v
AI-Assisted Notification Investigation

The goal is to make notification configuration easier to explore without manually piecing together information from multiple resources.

Key engineering features

1. Notification resource ingestion

The connector is designed to ingest four categories of resources.

Workflows

Workflow metadata describes notification behavior. Depending on the fields exposed by the connector, this might include names, descriptions, active status, tags, and delivery-channel steps such as email, SMS, push, in-app, and chat.

Topics

Topic metadata helps identify notification categories and their associated subscriber information, where available.

Notification groups

Group metadata provides organizational context for workflows and notification categories.

Provider integrations

Integration metadata describes configured delivery providers, channel types, and relevant status information.

Together, these resources provide a foundation for exploring how notification workflows relate to delivery infrastructure.

The exact fields and relationships depend on the Novu API and the connector's current implementation.

2. Deterministic change detection

Notification configurations change as teams add channels, modify workflows, or update provider settings.

The connector uses canonical SHA-256 fingerprints to identify meaningful changes in supported records.

The intended process is:

1. Retrieve a resource from Novu.
2. Normalize its relevant fields.
3. Calculate a deterministic fingerprint.
4. Compare it with the previously synchronized version.
5. Skip unchanged records or process updated records.

This approach helps reduce unnecessary reprocessing during repeated synchronization.

The fingerprint must include the fields that matter for synchronization. A field excluded from the fingerprint will not independently trigger an update.

3. Deletion detection and reconciliation

Notification workflows, topics, and integrations are not permanent. When a resource is removed upstream, its downstream representation needs a cleanup strategy.

The connector is designed to emit tombstones such as:

{"_deleted": True}

These records communicate deletion to the downstream pipeline.

The full cleanup process depends on the configured ingestion and cleanup mechanisms. Emitting a tombstone does not, by itself, guarantee removal from every relational, graph, and vector store.

4. Credential-safe ingestion

Notification providers often require credentials, including email API keys, SMS authentication tokens, and chat webhook secrets.

The connector describes a redaction strategy intended to prevent sensitive credential values from entering Cognee documents.

For example, a provider integration might be represented as:

Type: Novu Integration Provider
Provider ID: sendgrid
Channel Type: email
Status: Active
Credentials: [Redacted & Protected]

This retains useful information about the integration without deliberately copying its credentials into memory.

Redaction must apply consistently to every supported resource and data path. URLs, custom headers, nested credential fields, and diagnostic logs also need careful handling.

2. Installation and Prerequisites

The connector is distributed as a community package within the Cognee Community repository.

Prerequisites

• Python 3.10 or a version supported by the package.
• A Novu API key with permissions for the resources being ingested.
• Cognee and the connector's dependencies.
• Access to the Cognee Community repository.

Step 1: Clone the repository

git clone https://github.com/topoteretes/cognee-community.git
cd cognee-community

Step 2: Install the connector

Using uv:

uv pip install -e packages/connector/novu

Run this command inside an appropriately configured Python environment.

Check the package metadata and repository instructions for the supported Python and dependency versions.

Step 3: Configure your API key

Set your Novu API key through an environment variable:

export NOVU_API_KEY="your-novu-api-key"

Keep credentials out of source control. For production deployments, use a secrets manager or another approved credential-management mechanism.

3. Quickstart: Ingesting Notification Metadata into Cognee

The following example illustrates the intended workflow: configure the connector, ingest notification resources, process the dataset, and query the resulting memory.

Verify the current connector and Cognee API signatures against the repository before running the example.

import asyncio
import os

import cognee

from cognee_community_connector_novu import (
    novu,
    NOVU_REMEMBER_KWARGS,
)


async def main():
    api_key = os.environ["NOVU_API_KEY"]

    source = novu(
        api_key=api_key,
        include=(
            "workflows",
            "topics",
            "notification_groups",
            "integrations",
        ),
    )

    dataset_name = "notification_infrastructure"

    print("Ingesting Novu notification infrastructure...")

    await cognee.remember(
        source,
        dataset_name=dataset_name,
        **NOVU_REMEMBER_KWARGS,
    )

    print("Processing the knowledge graph...")

    await cognee.cognify(dataset_name=dataset_name)

    query = (
        "Which workflows send alerts through SMS or chat, "
        "and which delivery providers are active?"
    )

    print(f"\nQuerying: {query}")

    results = await cognee.search(
        search_text=query,
        dataset_name=dataset_name,
    )

    for index, match in enumerate(results, start=1):
        print(f"\n[{index}] {match}")


if __name__ == "__main__":
    asyncio.run(main())

What does this workflow do?

Configure the source. The connector reads the API key from the environment and selects the supported resource types.

Ingest the metadata. cognee.remember() sends the selected resources through the configured ingestion workflow.

Process the memory. cognee.cognify() performs the graph-processing step for the dataset.

Query the notification architecture. The search asks which workflows use SMS or chat and which providers are active.

The results depend on the resource fields ingested, the relationships extracted, and the installed versions of the connector and Cognee.

Example questions to explore

After ingestion and memory processing complete successfully, try questions such as:

• Which workflows include SMS delivery steps?
• Which workflows use chat notifications?
• Which provider integrations are marked active?
• Which topics have the largest subscriber counts, if those counts are available?
• Which workflows reference a particular notification group?

These examples illustrate potential use cases. Each query requires the corresponding information to exist in the ingested records or extracted graph.

4. Entity Types and Representation

The connector organizes notification resources into structured documents for downstream processing.

Entity type	Example document title	Useful context
Workflow	Novu Workflow: Password Reset [Active]	Workflow ID, name, description, steps, tags, status
Topic	Novu Topic: Security Alerts	Topic key, name, available subscriber information
Group	Novu Group: Authentication	Group ID and name
Integration	Novu Integration: Sendgrid (email) [Active]	Provider ID, channel type, status, primary configuration

These examples illustrate a possible representation. Actual document titles and available fields depend on the connector implementation.

A shared representation makes it easier to search for related notification resources and explore their relationships.

For example, an engineer investigating a password-reset notification might want to identify the workflow, inspect its delivery channels, and determine which provider handles the relevant integration.

Whether those connections are available in a useful form depends on the metadata ingested and the relationships created during graph processing.

5. Security: Protecting Provider Credentials

Notification integrations often store credentials required to send email, SMS, push notifications, or chat messages.

Consider a provider payload containing an API key:

# Illustrative payload only. Never use real credentials in examples.
{
    "providerId": "sendgrid",
    "credentials": {
        "apiKey": "example-secret",
    },
}

The intended memory representation excludes the sensitive value:

Type: Novu Integration Provider
Provider ID: sendgrid
Channel Type: email
Credentials: [Redacted & Protected]

A robust implementation should:

• Redact credentials before generating documents.
• Handle nested credential fields consistently.
• Avoid exposing secrets in debug logs or error messages.
• Review webhook URLs and custom headers for embedded tokens.
• Test redaction across every supported resource type.
• Restrict access to the resulting Cognee dataset.

These controls reduce exposure, but a redaction policy should not be described as an absolute guarantee without reviewing and testing the full implementation.

6. Offline Testing and Verification

The project describes a 38-test offline suite that uses an in-memory mock client rather than making live external HTTP requests.

The reported coverage includes resource ingestion, synchronization, credential redaction, and deletion handling.

Run the test suite with:

pytest packages/connector/novu/tests -v

Check linting:

python -m ruff check packages/connector/novu

Check formatting:

python -m ruff format --check packages/connector/novu

The reported result is:

38 tests passed in under 35 seconds

Treat this as the project's reported result until you reproduce it against the relevant commit.

Offline tests provide repeatable coverage of the behaviors they exercise. They do not replace integration testing against Novu or verification of the complete downstream memory-processing and cleanup workflow.

7. Engineering Lessons

This connector highlights several principles for building reliable AI integrations.

Model relationships, not just records. Workflow, topic, group, and integration metadata become more useful when their relationships are available to the retrieval system.

Track meaningful changes. Deterministic fingerprints help identify updates and reduce unnecessary reprocessing.

Treat deletion as an explicit workflow. Upstream removal does not automatically erase downstream representations. Tombstones and cleanup operations need separate verification.

Minimize sensitive data. Store the information needed for retrieval without copying provider credentials into documents or indexes.

Keep operational claims grounded in source data. Knowing that a provider is configured does not prove successful message delivery. Delivery outcomes require appropriate delivery-status or event data.

These principles apply to notification systems, communication platforms, event-driven applications, and other infrastructure managed through APIs.

8. Project Links

Explore the connector and its implementation in the Cognee Community repository.

The connector package is located at:

packages/connector/novu

Review the source code and tests for current API support, available resource fields, credential handling, and synchronization behavior.

Final Thoughts

Connecting Novu to Cognee offers a path toward making notification infrastructure easier to explore through natural-language queries.

The potential value comes from connecting workflow definitions, delivery channels, topics, and provider configurations in one searchable memory system.

The difficult engineering work is keeping this context accurate, secure, and synchronized as the application evolves.

If you were building an AI agent for notification infrastructure, what would you integrate next: delivery events, subscriber preferences, message templates, or incident-management data?

Share your thoughts in the comments.