DEV Community: Deshu

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

Deshu — Tue, 20 Oct 2026 14:32:28 +0000

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:

Retrieve a resource from Novu.
Normalize its relevant fields.
Calculate a deterministic fingerprint.
Compare it with the previously synchronized version.
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.

Mapping Distributed Webhook Architectures and Event Schemas into Cognee with Svix

Deshu — Tue, 20 Oct 2026 14:29:50 +0000

Modern cloud applications rely on asynchronous events to communicate across services. Payments, user registrations, subscription changes, and background jobs often trigger webhooks that connect applications to external systems.

Platforms such as Svix help developers manage webhook applications, endpoint subscriptions, and event delivery infrastructure.

But as an event-driven architecture grows, understanding its topology becomes harder.

Engineers might need to answer questions such as:

Which endpoints subscribe to invoice.payment_failed?
Which applications have disabled webhook endpoints?
What is the potential impact of changing the user.created event schema?
Which endpoints are configured to receive billing events?

Finding these answers often requires switching between dashboards, API responses, and application documentation.

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

That's the idea behind cognee-community-connector-svix, a community connector designed to ingest Svix webhook infrastructure into Cognee using dlt.

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

1. Architecture: From Webhook Infrastructure to AI Memory

The connector lives in the Cognee Community repository under:

packages/connector/svix

Its intended architecture connects three components:

Svix: Provides webhook application, endpoint, and event-type metadata.
dlt: Handles structured ingestion and synchronization.
Cognee: Processes ingested information for knowledge-graph and semantic retrieval.

The data flow looks like this:

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

The goal is to make webhook relationships easier to investigate without manually reconstructing the topology from multiple resources.

Key engineering features

1. Application, endpoint, and event-type ingestion

The connector is designed to ingest three resource categories.

Applications

Application metadata provides context about the services or tenants associated with webhook infrastructure. Depending on the source fields, this might include application IDs, names, UIDs, creation timestamps, and rate-limit settings.

Endpoints

Endpoint metadata describes webhook destinations and their configuration. Relevant fields might include target URLs, descriptions, versions, subscription settings, and active or disabled status.

Event types

Event-type metadata describes the events applications produce, including their names, descriptions, status, and available schema information.

Together, these resources provide a foundation for exploring how event producers and consumers relate to each other.

The exact fields available depend on the Svix API responses and the connector's implementation.

2. Deterministic change detection

Webhook configurations evolve. Endpoints get disabled, subscriptions change, and event definitions are updated.

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

The intended process is:

Retrieve a resource from Svix.
Normalize its relevant fields.
Compute a deterministic fingerprint.
Compare the result with the previously synchronized version.
Skip unchanged records or process updated ones.

This approach helps reduce unnecessary reprocessing during repeated synchronization.

Fingerprinting works only for fields included in the fingerprint calculation. Those fields should reflect the changes that matter to downstream memory.

3. Deletion detection and reconciliation

An application or endpoint removed from Svix should not remain indefinitely in downstream memory.

The connector is designed to emit tombstones for deleted resources:

{"_deleted": True}

These records communicate deletion to the downstream pipeline.

The full cleanup process still depends on the configured ingestion and cleanup mechanisms. Tombstone emission alone does not guarantee that every corresponding record has been removed from relational, graph, and vector storage.

4. Secret-safe ingestion

Webhook infrastructure contains sensitive configuration, including signing secrets and custom authorization headers.

The connector describes a redaction policy intended to replace sensitive values before documents enter Cognee memory.

For example:

Endpoint Secret: [Redacted & Protected]
Custom Headers: [Redacted & Protected]

This preserves useful information about the configuration while avoiding unnecessary exposure of credentials.

Redaction must cover every relevant ingestion path. Target URLs also require care because query parameters sometimes contain tokens or other sensitive values.

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 current package.
A Svix API token with the permissions required 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/svix

Run this command inside an appropriately configured Python environment.

Check the connector's package metadata and installation instructions for its supported Python and dependency versions.

Step 3: Configure your API token

Set your Svix API token as an environment variable:

export SVIX_API_TOKEN="your-svix-api-token"

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

3. Quickstart: Ingesting Webhook Topology into Cognee

The following example illustrates the intended integration: configure the source, ingest the resources, process the dataset, and query the resulting memory.

Verify the connector and Cognee API signatures against the current implementation before running it.

import asyncio
import os

import cognee

from cognee_community_connector_svix import (
    svix,
    SVIX_REMEMBER_KWARGS,
)


async def main():
    api_token = os.environ["SVIX_API_TOKEN"]

    source = svix(
        api_token=api_token,
        include=(
            "applications",
            "endpoints",
            "event_types",
        ),
        include_archived_events=False,
    )

    dataset_name = "webhook_architecture"

    print("Ingesting Svix webhook infrastructure...")

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

    print("Processing knowledge graph...")

    await cognee.cognify(dataset_name=dataset_name)

    query = (
        "Which endpoints are subscribed to billing events, "
        "and are any of them currently disabled?"
    )

    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 the workflow do?

Configure the source. The connector reads the API token from the environment and selects applications, endpoints, and event 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 topology. The search asks which endpoints receive billing events and whether any are disabled.

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

Example questions to explore

After the ingestion pipeline completes successfully, try questions such as:

Which endpoints are associated with billing events?
Which endpoint configurations are disabled?
Which event types are marked as archived?
Which applications have configured rate limits?
What endpoint metadata is available for a particular event type?

These examples demonstrate potential use cases. The connector must ingest the relevant fields and preserve the necessary relationships for each query to produce reliable answers.

4. Entity Types and Representation

The connector organizes webhook resources into structured documents suitable for downstream processing.

Entity type	Example document title	Useful context
Application	`Svix App: Billing Service`	Application ID, UID, creation time, available rate-limit metadata
Endpoint	`Svix Endpoint: Stripe Webhook [Active]`	Target URL, description, version, subscription metadata, status
Event type	`Svix Event: invoice.paid [Active]`	Event name, description, status, available schema information

These titles illustrate a possible representation. Actual titles and extracted properties depend on the implementation.

The broader goal is to make related resources easier to find and connect.

For example, an engineer investigating a billing event might want to identify the relevant application, inspect its endpoint configurations, and review the event's available schema metadata.

Whether those connections are queryable depends on the records ingested and the graph relationships created during processing.

5. Security: Protecting Webhook Signing Secrets

Svix webhook signing secrets are used by consumers to verify webhook signatures. Exposing them in a knowledge graph or vector store would create unnecessary risk.

Consider an endpoint payload containing a signing secret:

# Illustrative payload only. Never use real credentials in examples.
{
    "url": "https://api.partner.com/webhook",
    "secret": "example-signing-secret",
}

The intended memory representation excludes the secret:

Type: Svix Webhook Endpoint
Target URL: https://api.partner.com/webhook
Endpoint Secret: [Redacted & Protected]
Custom Headers: [Redacted & Protected]

A robust implementation should:

Remove signing secrets before document generation.
Avoid indexing sensitive custom-header values.
Review target URLs for embedded credentials or sensitive query parameters.
Keep API tokens outside source control.
Test redaction across every supported resource type.
Check generated documents and logs for accidental exposure.

Redaction should happen before sensitive values enter the downstream document and embedding pipeline.

These controls reduce risk, but they should not be described as an absolute guarantee unless the full implementation has been reviewed and tested.

6. Offline Testing and Verification

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

The reported test areas include ingestion behavior, resource handling, change detection, deletion signals, and secret redaction.

Run the test suite with:

pytest packages/connector/svix/tests -v

Check linting:

python -m ruff check packages/connector/svix

Check formatting:

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

The reported result is:

39 tests passed in under 25 seconds

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

Offline tests provide repeatable coverage of the scenarios they exercise. They do not replace integration tests against the Svix API or verification of the complete downstream deletion and memory-processing workflow.

7. Engineering Lessons

This connector highlights several principles that apply to AI integrations for event-driven systems.

Model relationships, not just records. Application, endpoint, and event metadata become more useful when their relationships are available to the retrieval system.

Track meaningful changes. Deterministic fingerprints help identify updates without repeatedly processing unchanged records.

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

Minimize sensitive data. Signing secrets and authorization values should never be included in memory documents when they are unnecessary for retrieval.

Keep schema claims grounded. Understanding the impact of an event-schema change requires the relevant schema versions and consumer relationships. A list of event names alone is insufficient to determine the full blast radius.

These principles apply to webhook platforms, event brokers, message queues, and other distributed integration systems.

8. Project Links

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

The connector package is located at:

packages/connector/svix

Review the source code and tests for the current API support, resource fields, redaction rules, and synchronization behavior.

Final Thoughts

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

The value comes from combining application metadata, endpoint configuration, event definitions, and reliable synchronization in one searchable memory system.

The difficult engineering work is ensuring that this context remains accurate, useful, and secure as the underlying architecture changes.

If you were building an AI agent for distributed systems, which integration would you add next: message queues, event brokers, API gateways, or distributed tracing?

Share your thoughts in the comments.

Ingesting Real-Time Website Uptime and Incident Telemetry into Cognee with UptimeRobot

Deshu — Tue, 20 Oct 2026 14:27:01 +0000

When a production service goes down, site reliability engineers (SREs), incident commanders, and on-call responders need answers quickly.

They need to know which monitors are failing, whether maintenance explains the disruption, which alert contacts are configured, and how reliable the affected service has been over time.

Questions often include:

Which monitored endpoints are currently down?
Is the outage related to a scheduled maintenance window?
Which alert contacts are configured for the affected monitors?
What historical uptime information is available for the degraded service?

Monitoring dashboards provide essential operational data, but investigating an incident often means switching between multiple pages and systems.

What if an AI agent could retrieve relevant monitoring metadata from a knowledge graph and use it to help investigate these questions?

That's the idea behind cognee-community-connector-uptimerobot, a community connector designed to bring UptimeRobot monitoring data into Cognee using dlt.

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

1. Architecture: Bringing UptimeRobot Data into AI Memory

The UptimeRobot connector integrates the monitoring platform's API with Cognee's data ingestion and memory-processing workflow.

The intended architecture is:

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

The connector is designed to ingest four categories of monitoring resources:

Monitors: Endpoint monitoring configuration and available status information.
Public status pages: Public-facing service status metadata.
Alert contacts: Configured notification contacts and channel information.
Maintenance windows: Scheduled maintenance configuration.

The connector uses UptimeRobot API v2 endpoints, including /getMonitors, /getPSPs, /getAlertContacts, and /getMWindows, according to the described implementation.

The exact fields and resource coverage depend on the API responses and the connector's current implementation.

Key engineering features

1. Deterministic change detection

Repeatedly processing unchanged monitoring records wastes resources and introduces unnecessary updates.

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

The intended behavior is straightforward:

Retrieve the upstream resource.
Normalize its relevant fields.
Calculate a deterministic fingerprint.
Compare it with the previously synchronized version.
Skip unnecessary updates or process changed records.

For example, a monitor's configuration might remain unchanged for several synchronization cycles. Fingerprinting helps the connector identify records that do not need reprocessing.

A meaningful change in a monitored resource should produce a different fingerprint, provided the changed field is included in the fingerprint calculation.

2. Deletion detection and reconciliation

Monitoring resources are not permanent. A monitor might be removed, an alert contact might be deleted, or a status page might be retired.

The connector is designed to emit tombstones for deleted resources:

{"_deleted": True}

These records communicate deletion to the downstream pipeline.

The complete cleanup process still depends on the configured merge behavior and Cognee's downstream cleanup mechanism. Tombstone generation alone does not guarantee that every stale record disappears from relational, graph, and vector storage.

3. Sensitive-field redaction

Monitoring configurations sometimes contain HTTP Basic Authentication credentials or webhook URLs with sensitive query parameters.

The connector describes a redaction strategy that replaces sensitive values before generating memory documents.

For example, an authenticated monitor might be represented as:

Type: HTTP Monitor
Name: Internal Vault API
Target URL: https://vault.internal.net/health
HTTP Authentication: [Redacted & Protected]

The goal is to retain useful operational context without exposing authentication values.

Webhook URLs require particular care because secrets might appear in their paths or query parameters. A robust implementation should redact sensitive components rather than assume every URL is safe to store.

The redaction policy should be tested across every supported resource type and data path before production use.

2. Installation and Prerequisites

The connector is distributed within the Cognee Community repository.

Prerequisites

Python 3.10 or a version supported by the current package.
An UptimeRobot API key with the permissions required for the selected resources.
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/uptimerobot

Run this inside an environment where the required Python dependencies are installed.

Check the connector's package metadata and repository instructions for the supported Python version and dependency setup.

Step 3: Configure the API key

Set your UptimeRobot API key through an environment variable:

export UPTIMEROBOT_API_KEY="your-uptimerobot-api-key"

Use a suitable key for your environment and avoid committing credentials to source control.

The connector's supported key types and required permissions should be confirmed against the current UptimeRobot API and implementation.

3. Quickstart: Ingesting Monitoring Data into Cognee

The following example demonstrates the intended workflow: configure the source, ingest monitoring resources, process the dataset, and ask an operational question.

Verify the connector's current API signatures before running the example.

import asyncio
import os

import cognee

from cognee_community_connector_uptimerobot import (
    uptimerobot,
    UPTIMEROBOT_REMEMBER_KWARGS,
)


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

    source = uptimerobot(
        api_key=api_key,
        include=(
            "monitors",
            "public_status_pages",
            "alert_contacts",
            "maintenance_windows",
        ),
        only_down=False,
    )

    dataset_name = "production_infrastructure"

    print("Ingesting UptimeRobot telemetry into Cognee...")

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

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

    await cognee.cognify(dataset_name=dataset_name)

    query = (
        "List the monitors that are currently down, "
        "their available historical uptime metrics, "
        "and their configured alert contacts."
    )

    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?

First, it configures the source. The connector reads the API key from the environment and selects the monitoring resource types to ingest.

Next, it ingests the data. cognee.remember() passes the source through the configured ingestion workflow.

Then, it processes the memory. cognee.cognify() performs the graph-processing step for the selected dataset.

Finally, it runs a query. The search asks for current monitor status, available historical uptime information, and configured alert contacts.

The exact behavior of the search API depends on the installed Cognee version. Confirm the supported arguments and result structure before deploying the script.

Also, ingestion does not guarantee that every requested metric is available. The query can return only information that the API exposes, the connector ingests, and the memory pipeline successfully processes.

4. Entity Types and Representation

The connector organizes monitoring resources into human-readable documents that are intended for embedding and entity extraction.

Entity type	Example document title	Potentially useful context
Monitor	`Monitor: Auth Gateway [UP]`	Status, target, port, interval, available uptime metrics
Public status page	`Status Page: Global Status [Active]`	Page URL, custom domain, associated monitors
Alert contact	`Alert Contact: DevOps Pager`	Contact type, configured status
Maintenance window	`Maintenance Window: Weekly DB Backup`	Schedule, duration, recurrence configuration

The examples above illustrate a possible representation. The exact field names and title formats depend on the implementation.

This structured context helps an AI agent connect related information across monitoring resources.

For example, an agent might retrieve a monitor and its configured alert contact in the same investigation. Whether the relationship is available in a useful form depends on the records ingested and the graph-processing behavior.

5. Security: Protecting Credentials and Webhook Secrets

Monitoring integrations often handle data that should not be exposed to downstream memory systems.

Consider a monitor configuration containing HTTP authentication:

# Illustrative source payload, not a real credential.
{
    "friendly_name": "Internal Vault API",
    "http_username": "example-user",
    "http_password": "example-password",
}

The intended memory representation excludes the authentication values:

Type: HTTP Monitor
Name: Internal Vault API
HTTP Authentication: [Redacted & Protected]

This approach preserves useful context about the configuration without storing the credentials themselves.

Webhook URLs need similar treatment. A URL might contain a token in its path or query string, so the connector should avoid indexing the raw value when it contains sensitive information.

For production deployments, consider these safeguards:

Keep API keys in environment variables or a secrets manager.
Test redaction for every resource type.
Check generated documents and logs for accidental credential exposure.
Review URL handling, including query parameters and embedded credentials.
Restrict access to the resulting Cognee dataset.
Rotate credentials if exposure is suspected.

Redaction tests increase confidence in the intended behavior. They do not prove that secrets are impossible to expose through every untested code path.

6. Offline Testing and Verification

The project describes a 43-test offline unit and integration test suite that uses a mock client instead of relying on live external API calls.

The reported coverage includes API behavior, data normalization, redaction, state synchronization, and deletion handling.

Run the connector's tests with:

pytest packages/connector/uptimerobot/tests -v

Run the lint checks:

python -m ruff check packages/connector/uptimerobot

Verify formatting:

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

The reported result is:

43 tests passed in under 25 seconds

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

Offline tests help make connector behavior repeatable without network dependencies. They do not replace integration tests against the actual UptimeRobot API or validation of the complete Cognee ingestion and cleanup pipeline.

7. Engineering Lessons

This connector highlights several principles that apply to AI-powered observability integrations.

Synchronize changes, not just snapshots. Monitoring metadata evolves. Deterministic fingerprints help identify changed records and reduce redundant processing.

Treat deletion as a separate workflow. A deleted source record does not automatically disappear from downstream storage. Tombstones and cleanup operations require explicit verification.

Minimize sensitive data. Ingest the information needed for useful retrieval without unnecessarily copying credentials into memory.

Keep metrics grounded in source data. Historical uptime calculations require the appropriate source fields and reporting period. Do not infer a 30-day uptime ratio when the underlying data is unavailable.

Keep live monitoring authoritative. Knowledge graphs provide context, but time-sensitive incident decisions should be validated against the monitoring system.

Together, these principles help make AI-assisted incident investigation more useful and more trustworthy.

8. Project and Repository

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

The connector package is located at:

packages/connector/uptimerobot

Review the source code and tests for the current API support, supported resource types, redaction behavior, and deletion-handling implementation.

Final Thoughts

Connecting UptimeRobot to Cognee creates a path toward AI-assisted investigations grounded in monitoring metadata.

The potential value comes from bringing related information into one searchable memory system: monitor configuration, service status, alert contacts, and maintenance schedules.

The engineering challenge is to keep that information secure, current, and consistent with the source system.

If you were building an AI incident investigation agent, what would you integrate next: application logs, distributed traces, deployment history, or incident-management data?

Share your thoughts in the comments.

Proactive Incident Intelligence: Syncing Checkly Monitoring into Cognee Memory with dlt

Deshu — Tue, 20 Oct 2026 12:59:04 +0000

Modern applications cannot afford to wait for users to report outages. Engineering teams use synthetic monitoring platforms like Checkly to test REST APIs, execute automated Playwright browser flows, and detect failures before they affect more users.

But detecting an incident is only half the problem. Engineers also need context to understand what is failing, which alerts are configured, and whether maintenance windows explain the behavior.

During an incident, an engineer or AI agent might need to answer questions such as:

Which synthetic API checks or checkout flows are failing?
Which alert channels notify the team about API degradation?
Are the affected endpoints covered by a scheduled maintenance window?

Answering these questions often requires information from multiple resources.

In this article, we explore cognee-community-connector-checkly, a Checkly data-source connector developed for the Mergetober 2026 hackathon. It is designed to synchronize synthetic monitoring metadata into Cognee using dlt.

We'll examine its architecture, secret-safe ingestion strategy, health-status synchronization, deletion handling, and offline testing approach.

1. Architecture: Bringing Synthetic Monitoring into AI Memory

The connector resides in the Cognee Community monorepo, under:

packages/connector/checkly

Its architecture connects three components:

Checkly: Supplies synthetic monitoring data and configuration.
dlt: Handles structured data ingestion and incremental synchronization.
Cognee: Processes ingested information for knowledge-graph and semantic retrieval.

The intended data flow is:

Checkly API
    |
    v
Checkly Data-Source Connector
    |
    v
dlt Ingestion Pipeline
    |
    v
Cognee Memory Processing
    |
    v
AI-Powered Incident Investigation

The goal is to give AI agents useful operational context without treating monitoring data as static documentation.

Core design principles

1. Secret-safe metadata ingestion

Synthetic checks often use authenticated requests, API tokens, cookies, and basic authentication. Check groups might also contain environment variables with sensitive values.

Indexing these values in a knowledge graph or vector store introduces unnecessary risk.

The connector follows a key-only redaction strategy for supported headers and environment variables. Instead of storing a sensitive header value, the intended representation retains its name:

Headers (keys only): Authorization

Likewise, environment variables are represented by their names:

Environment Variables (keys only):
STRIPE_SECRET_KEY, API_BASE_URL

This preserves structural information that helps explain a check's configuration without deliberately copying the associated values into memory.

The implementation must apply this policy consistently across every ingestion path. Redaction tests help verify the behavior, but they do not replace a broader security review.

2. Health-status propagation

Check definitions describe what should be monitored. Health-status data helps explain what is happening now.

The connector is designed to retrieve check-status information through the /v1/check-statuses endpoint and associate the results with relevant monitoring records.

The intended benefit is to make health information available alongside check configuration. An AI agent could then investigate questions such as:

Which synthetic API checks are failing or degraded, and which alert channels are configured to notify the team?

The usefulness of these answers depends on the connector's status mapping, the API response, and the freshness of the latest successful synchronization.

3. Merge-based synchronization and deletion handling

Monitoring resources change over time. Checks are updated, groups are removed, and temporary configurations become obsolete.

The connector uses merge-based ingestion:

write_disposition="merge"

It also uses a _deleted tombstone mechanism with hard-delete semantics to represent upstream deletions.

The intended workflow is:

Detect that a resource no longer exists upstream.
Emit a tombstone for the deleted resource.
Pass the deletion signal into downstream cleanup.
Remove stale records from the relevant storage layers.

Cognee's cleanup process must execute successfully for the deletion to propagate through relational, graph, and vector storage. Emitting a tombstone alone does not guarantee complete cleanup.

2. Quickstart: Integrating Checkly with Cognee

Step 1: Install the dependencies

Install the connector and its main dependency:

pip install cognee cognee-community-connector-checkly

Use a virtual environment and verify the supported package versions in the repository before running the integration.

Step 2: Configure your Checkly API key

Retrieve your API key from Checkly and configure it through an environment variable.

For Linux or macOS:

export CHECKLY_API_KEY="your-checkly-api-key"

For an account-specific configuration, set the optional account ID:

export CHECKLY_ACCOUNT_ID="your-checkly-account-id"

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

Step 3: Synchronize monitoring metadata

The following example illustrates the intended integration. Verify the current connector and Cognee API signatures against the repository before running it.

import asyncio
import os

import cognee
from cognee_community_connector_checkly import checkly_source


async def main():
    api_key = os.environ["CHECKLY_API_KEY"]
    account_id = os.environ.get("CHECKLY_ACCOUNT_ID")

    source = checkly_source(
        api_key=api_key,
        account_id=account_id,
        resource_name="checkly_prod",
        include=(
            "checks",
            "check_groups",
            "alert_channels",
            "maintenance_windows",
        ),
    )

    print("--- Ingesting Checkly metadata ---")

    await cognee.remember(
        source,
        dataset_name="synthetic_monitoring_memory",
        primary_key="id",
        write_disposition="merge",
        max_rows_per_table=0,
    )

    print("Sync stats:", source.cognee_sync_stats)

    print("\n--- Querying monitoring memory ---")

    result = await cognee.search(
        query_text=(
            "Which synthetic API checks are failing or degraded, "
            "and which alert channels notify on failure?"
        ),
        query_type=cognee.SearchType.GRAPH_COMPLETION,
        datasets=["synthetic_monitoring_memory"],
    )

    print("Monitoring insights:\n", result)


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

This example demonstrates three main operations:

Configure the Checkly connector using environment variables.
Ingest checks, groups, alert channels, and maintenance-window metadata.
Query the resulting Cognee dataset using natural language.

The exact resources available depend on the connector implementation and the installed version. Confirm that each resource type is supported before using the example in production.

Step 4: Ask incident-related questions

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

Which synthetic API checks are failing?
Which alert channels are configured for the affected checks?
Which check groups contain the failing checks?
Which maintenance windows overlap with the incident?
What monitoring configuration is associated with the affected endpoint?

These questions illustrate how structured monitoring metadata might support AI-assisted incident investigation.

One important distinction: synchronizing monitoring metadata does not automatically create a real-time incident response system. The freshness of the results depends on synchronization frequency, and critical incident decisions should be checked against authoritative monitoring data.

3. Comprehensive Offline Test Suite

Reliable ingestion requires more than successful API responses. A connector must handle network failures, rate limits, changing health states, sensitive values, and deleted resources.

The project reports a 45-test offline suite covering the following areas.

API client behavior

Retry handling for HTTP 502 responses.
Connection-drop retries.
Rate-limit header parsing.
Account scoping through X-Checkly-Account.
API token redaction.

Secret-redaction verification

Tests use sensitive-looking values in fixtures to verify that credentials do not appear in emitted document text.

Examples include:

Authorization: Bearer secret
STRIPE_SECRET_KEY

The important distinction is between retaining a configuration key and retaining its sensitive value. The connector should preserve only the information required for its intended use.

Health-transition detection

A synthetic check might report HEALTHY during one sync and FAILING during the next.

The connector's fingerprinting strategy is designed to detect meaningful changes and re-emit updated records. This helps prevent AI memory from remaining stuck on an earlier state.

Forget-on-delete verification

When a Checkly resource is deleted, the connector should detect the removal and emit a _deleted=True tombstone.

The downstream pipeline must then process that tombstone and remove the corresponding stale data.

SQLite-backed dlt pipeline

The suite also describes an SQLite-backed pipeline that exercises merge behavior and hard-deletion handling across successive runs.

Run the tests with:

pytest packages/connector/checkly/tests -v

The reported result is:

45 passed in 22.70s

This is the reported test result, not an independently verified result. Reproduce it against the relevant commit before presenting it as your own verified run.

Offline tests provide repeatable coverage of the behaviors they exercise. They do not replace integration testing against Checkly or validation of the complete downstream cleanup pipeline.

4. Engineering Lessons

This connector illustrates several principles that apply to AI integrations across monitoring and infrastructure systems.

Minimize sensitive data. Retain the metadata needed for retrieval without unnecessarily copying credentials into documents or indexes.

Track changing operational states. Monitoring data evolves. Incremental synchronization should recognize meaningful changes rather than treating every record as immutable.

Design deletion handling explicitly. Removing a resource from its source does not automatically remove every downstream representation. Tombstones and cleanup operations need separate verification.

Make synchronization observable. Fingerprints and sync statistics help engineers understand which records changed and which were skipped.

Keep operational truth authoritative. AI memory helps connect related information, but live incident status should be verified against the monitoring system when decisions depend on current conditions.

These principles apply to other integrations involving application logs, distributed traces, deployment systems, infrastructure inventories, and incident-management platforms.

5. Project Links

The connector was developed as part of the Mergetober 2026 hackathon.

Pull request: Cognee Community PR #447
Feature issue: Cognee Issue #6124
Repository: Cognee Community Monorepo
Cognee: GitHub repository

Visit the pull request and repository for the implementation details, supported configuration, and current test results.

Final Thoughts

Connecting Checkly synthetic monitoring to Cognee offers a way to give AI agents more context about application reliability, monitoring configuration, and incident investigation.

The difficult engineering work lies in maintaining that context safely and accurately: protecting credentials, tracking health changes, synchronizing updates, and removing deleted resources.

If you were building an AI incident investigation agent, which data source would you integrate next: application logs, distributed traces, deployment history, or incident-management data?

Share your thoughts in the comments.

Building AI Agent Memory for Prefect Workflows with Cognee and dlt

Deshu — Tue, 20 Oct 2026 12:52:03 +0000

Building an Intelligent Workflow Brain: Syncing Prefect Orchestration into Cognee Memory with DLT

Modern data engineering and orchestration platforms manage hundreds of pipelines, scheduled deployments, and thousands of execution runs. When a midnight ETL job crashes, an engineer is often left piecing together what happened across disconnected logs, API records, and dashboards.

What if your AI agents had instant, contextual memory of your entire orchestration topology?

In this guide, we dive into the newly created Prefect data-source connector for Cognee
(cognee-community-connector-prefect), built as part of the Mergetober 2026 hackathon. We'll explore how it seamlessly bridges Prefect 2/3 orchestration metadata into Cognee's knowledge graph and vector stores using dlt, while upholding production-grade security, incremental state transitions, and forget-on-delete capabilities.

The Challenge: Orchestration Blind Spots for AI Agents

LLM-driven assistants and autonomous DevOps agents are only as capable as the memory backing them. While general documents and database tables are easy to ingest, workflow orchestration data introduces three unique engineering hurdles:

State Dynamics: Unlike static files, a flow run is mutable. It starts as PENDING, transitions to RUNNING, and terminates as COMPLETED, FAILED, or CRASHED. A naive ingestion pipeline that only queries creation timestamps misses terminal state transitions.
Secret Parameter Leakage: Prefect deployments and flow runs routinely accept runtime parameters—such as database passwords, connection strings, and webhook tokens. Indexing raw parameters into vector databases or knowledge graphs poses an unacceptable security risk.
Graph Deletion & Drift: When obsolete flows or ephemeral testing deployments are deleted in Prefect, they must be cleanly removed from Cognee's graph and vector memory rather than lingering as orphaned hallucinations.

Architecture: How the Prefect Connector Solves This

The cognee-community-connector-prefect package lives in the topoteretes/cognee-community
monorepo under packages/connector/prefect.

Key Engineering Decisions:
Strict Zero-Secret Parameter Policy: The connector parses parameters on deployments and runs, strictly rendering parameter keys only (e.g. Parameters (keys only): batch_size, target_table), ensuring zero passwords or API keys are ever committed to memory.
Merge Disposition with Tombstones: Configured with write_disposition="merge" and columns={"_deleted": {"data_type": "bool", "hard_delete": True}}. When an entity is deleted upstream, a tombstone row is emitted, enabling Cognee's orphan_cleanup to purge it from relational, graph, and vector stores without wiping the dataset.
Deterministic SHA-256 Fingerprints: Each record is fingerprinted across its core operational metadata. When a flow run updates from RUNNING to COMPLETED, its fingerprint shifts, prompting Cognee to re-index the updated state while untouched historical runs are skipped (stats["skipped"]).

Step-by-Step Walkthrough Step 1: Install Dependencies bash pip install cognee cognee-community-connector-prefect Step 2: Configure Your Prefect Environment

For a self-hosted Prefect instance:

bash
export PREFECT_API_URL="http://localhost:4200/api"

For Prefect Cloud:

bash
export PREFECT_API_URL="https://api.prefect.cloud/api/accounts//workspaces/"
export PREFECT_API_KEY="pnu_..."
Step 3: Ingest Orchestration Metadata with cognee.remember
python
import asyncio
import os
import cognee
from cognee_community_connector_prefect import prefect_source
async def main():
base_url = os.environ.get("PREFECT_API_URL", "http://localhost:4200/api")
api_key = os.environ.get("PREFECT_API_KEY")
# Ingest flows, deployments, and execution runs into memory
source = prefect_source(
base_url=base_url,
api_key=api_key,
resource_name="prefect_main",
include=("flows", "deployments", "flow_runs"),
# Optional: filter by tags or execution states
flow_run_states=["COMPLETED", "FAILED", "CRASHED"],
)
print("--- Starting Prefect Ingestion ---")
await cognee.remember(
source,
dataset_name="orchestration_memory",
primary_key="id",
write_disposition="merge",
max_rows_per_table=0,
)
print("Sync Stats:", source.cognee_sync_stats)
# Query your workflow memory
print("\n--- Asking Cognee Memory ---")
result = await cognee.search(
query_text="What flows and deployments failed recently, and what are their schedules?",
query_type=cognee.SearchType.GRAPH_COMPLETION,
datasets=["orchestration_memory"],
)
print("Orchestration Insights:\n", result)
if name == "main":
asyncio.run(main())

Rigorous Offline Verification

To ensure rock-solid reliability without external network dependencies, the connector includes a 42-test offline suite:

Client Tests: Exponential backoff on HTTP 502/503, connection retry handling, rate-limit header parsing, and token redaction.
State Transition Tests: Verifies that a flow run moving from PENDING to FAILED is recognized and re-emitted on the subsequent sync.
Secret Redaction Tests: Verifies that test fixtures with sensitive passwords (db_password=supersecretpassword) never leak into document text.
Forget-on-Delete Tests: Confirms single-endpoint lookups detect deleted entities and issue tombstones.
Real SQLite DLT Pipeline: Verifies state survival and hard-deletion across consecutive pipeline runs.
bash
pytest packages/connector/prefect/tests -v
====================== 42 passed in 22.84s =======================

Summary & Links

Connecting workflow orchestration engines like Prefect into Cognee bridges the gap between infrastructure telemetry and cognitive AI agents. Operators can now query complex pipeline states naturally, conduct automated failure root-cause analysis, and build resilient autonomous workflows.

Pull Request: topoteretes/cognee-community#446
Feature Issue: topoteretes/cognee#6119
Repository: Cognee Community Monorepo