DEV Community: Pranjul Chaurasiya

Software Supply Chain Memory: Connecting JFrog Artifactory to Cognee GraphRAG

Pranjul Chaurasiya — Sat, 31 Oct 2026 17:57:32 +0000

In modern enterprise DevOps and platform engineering, the software supply chain is vast and complex. Binary artifact repositories like JFrog Artifactory store thousands of container images, language packages (npm, PyPI, Maven), and Helm charts alongside critical CI/CD build execution logs and SBOM dependencies.

When vulnerabilities arise (such as zero-day CVEs in upstream dependencies) or release rollbacks occur, platform engineers and security analysts need to rapidly answer connected questions:

Which Docker images deployed in staging were compiled from Git commit 9f8a2b?
What published modules declare dependencies on a specific vulnerable package version?
What is the SHA-256 cryptographic checksum and build timeline for our payment microservice release?

Because software supply chains are intrinsically hierarchical and relational (Repository $\rightarrow$ Artifact $\rightarrow$ Module $\rightarrow$ Build Run $\rightarrow$ Dependencies), flat text search often misses the full dependency context.

As part of the WeMakeDevs × Cognee Mergetober Hackathon, I developed and submitted the JFrog Artifactory data-source connector (cognee-community-connector-artifactory) for Cognee.

Here is an architectural deep dive into how we bridge Artifactory repository topologies and build provenance into Cognee's GraphRAG engine.

Architecture: Supply Chain Knowledge Graph

The connector extracts repository topologies, artifact metadata, and build provenance directly from Artifactory's REST and AQL (Artifactory Query Language) endpoints:

flowchart TD
    subgraph ArtifactoryAPI ["JFrog Artifactory REST & AQL"]
        A["/api/repositories (Local, Remote, Virtual)"]
        B["/api/search/aql (Artifact Metadata & Checksums)"]
        C["/api/build/{name}/{number} (Build Runs & SBOM)"]
    end

    subgraph ConnectorCore ["Security & Normalization Layer"]
        D["Pre-Ingestion Credential Scrubbing (Tokens & API Keys)"]
        E["Zero Code Execution (Passive JSON AST Parsing)"]
        F["Selective Scoping Filter (repo_keys, repo_type, ecosystems)"]
        G["Structured Markdown Document Serializer"]
    end

    subgraph CogneeEngine ["Cognee GraphRAG Ingestion"]
        H["dlt Source (DOCUMENT_SOURCE_ATTR='artifactory')"]
        I["Cognify Pipeline (LLM Relationship Extraction)"]
        J["Software Supply Chain Knowledge Graph (FalkorDB / NetworkX)"]
        K["Vector Store (Semantic Memory)"]
        L["write_disposition='replace' (Orphan Cleanup)"]
    end

    A & B & C --> D --> E --> F --> G --> H
    H --> I --> J & K
    H --> L

1. Multi-Resource Ingestion: Repositories, Artifacts, & Build Runs

The connector captures three synchronized levels of the software supply chain:

A. Repository Topologies (`artifactory_repositories`)

Ingests local, remote, and virtual repositories across container and package ecosystems (Docker, Maven, npm, PyPI, Helm, etc.), capturing package types, descriptions, repository layouts, and governance notes.

B. Cryptographic Artifact Integrity (`artifactory_artifacts`)

Serializes artifact coordinates, repository paths, file sizes, creation/modification timestamps, and cryptographic checksums:

# Artifactory Artifact: services/auth-service/v1.4.0/manifest.json
- **Repository:** docker-prod-local (docker / local)
- **File Size:** 2,048 bytes
- **Created At:** 2026-09-15T08:30:00Z
- **Checksums:**
  - `SHA-256`: `e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855`
  - `SHA-1`: `da39a3ee5e6b4b0d3255bfef95601890afd80709`
  - `MD5`: `d41d8cd98f00b204e9800998ecf8427e`

C. CI/CD Build Provenance & SBOM (`artifactory_builds`)

Captures CI/CD execution metadata, build agent details, VCS commit SHAs, published modules, and declared dependency trees for end-to-end provenance.

2. Invariants & Security Standards

To meet Cognee's strict maintainer standards, the connector enforces key architectural invariants:

Zero Code Execution

Ingesting binary artifact metadata must never risk code execution. The connector parses REST API responses and AQL structures strictly as passive JSON data.

Pre-Ingestion Credential Masking

Authentication headers (X-JFrog-Art-Api, Bearer access tokens, basic auth credentials) are scrubbed before any records reach Cognee memory:

def redact_credentials(text: str) -> str:
    for pattern, replacement in REDACTION_PATTERNS:
        text = pattern.sub(replacement, text)
    return text

"Forget-on-Delete" via Snapshot Sync

When obsolete packages are cleaned up by Artifactory retention policies, their graph nodes must not persist as ghost references in AI memory:

setattr(source, DOCUMENT_SOURCE_ATTR, "artifactory")

@dlt.resource(name="artifactory_artifacts", write_disposition="replace")
def artifacts_resource():
    ...

With write_disposition="replace", Cognee's native orphan cleanup prunes deleted packages on each resync while preserving active dependency links.

3. Querying Supply Chain Memory in Python

Connecting Artifactory to Cognee requires only a few lines of asynchronous code:

import asyncio
import os
import cognee
from cognee_community_connector_artifactory import artifactory_source

async def main():
    # 1. Initialize Artifactory source
    source = artifactory_source(
        base_url=os.environ.get("ARTIFACTORY_URL", "https://company.jfrog.io/artifactory"),
        token=os.environ["ARTIFACTORY_TOKEN"],
        repo_keys=["docker-prod-local", "libs-release-local"],
        include_artifacts=True,
        include_builds=True,
    )

    # 2. Add to Cognee memory and build Knowledge Graph
    print("Ingesting software supply chain metadata...")
    await cognee.add(source)

    print("Running Cognify: Constructing Supply Chain Knowledge Graph...")
    await cognee.cognify()

    # 3. Ask natural language supply chain questions
    query = "Which Docker image was built from commit 9f8a2b and what are its declared dependencies?"
    print(f"\nQuerying Cognee: '{query}'")

    results = await cognee.search(query)
    for res in results:
        print(f"\n[Supply Chain Insight]:\n{res}")

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

Verification & Code Quality

100% Offline Passing Tests: 18/18 unit tests pass offline with zero network calls (tests/test_artifactory.py).
Dependency Hygiene: Configured with flexible dependency ranges (cognee>=1.6.3,<2 and dlt>=1.9,<2) avoiding exact pins.
DCO Certified: Verified Developer Certificate of Origin (Signed-off-by).

Pull Request & Resources

Pull Request: topoteretes/cognee-community#601
Upstream Discussion: topoteretes/cognee#5728

How is your team indexing software supply chain artifacts and SBOM metadata for AI agents? Share your thoughts below!

Bridging the Lakehouse: Ingesting Databricks Unity Catalog into Cognee Knowledge Graphs

Pranjul Chaurasiya — Sat, 31 Oct 2026 17:13:49 +0000

In large enterprise lakehouses, finding the right dataset, understanding table schemas, and tracing ML model lineage is often a major hurdle for data scientists, product managers, and AI agents.

While catalog tools like Databricks Unity Catalog centralize table schemas, storage locations, and column-level comments, interacting with these assets via natural language requires a structured knowledge layer. Vector similarity alone fails when users ask questions that span organizational boundaries:

"Which production tables contain customer billing information, and what are their primary keys?"
"What AI/ML models are registered under champion status for credit risk prediction?"
"Show me all schemas in the analytics catalog that partition by event_date."

To solve this, I designed and implemented the official Databricks Unity Catalog data-source connector (cognee-community-connector-unitycatalog) for Cognee as part of the WeMakeDevs × Cognee Mergetober Hackathon.

Here is how we turn Unity Catalog metadata into a connected GraphRAG layer.

Architectural Blueprint

The connector traverses Unity Catalog’s three-level namespace hierarchy (catalog.schema.table) along with the AI/ML Model Registry, converting tabular definitions into rich semantic markdown nodes:

flowchart TD
    subgraph UnityCatalog ["Databricks Unity Catalog API"]
        A["/api/2.1/unity-catalog/catalogs"]
        B["/api/2.1/unity-catalog/schemas"]
        C["/api/2.1/unity-catalog/tables (Managed, External, Views)"]
        D["/api/2.1/unity-catalog/models (Registered AI Models)"]
    end

    subgraph ConnectorEngine ["Connector Ingestion & Normalization"]
        E["Selective Scope Filter (catalogs / schemas)"]
        F["Column Schema Serializer (Types, Comments, Partitions)"]
        G["Model Registry Metadata (Aliases, Runs, Tags)"]
        H["Credential Scrubbing (Databricks PAT & Bearer Redaction)"]
    end

    subgraph CogneeGraph ["Cognee GraphRAG Ecosystem"]
        I["dlt Source (DOCUMENT_SOURCE_ATTR='unitycatalog')"]
        J["Cognify Entity Extractor"]
        K["Lakehouse Knowledge Graph (FalkorDB / NetworkX)"]
        L["Vector Index (Qdrant)"]
        M["Forget-on-Delete Orphan Pruner"]
    end

    A & B & C & D --> E --> F & G --> H --> I
    I --> J --> K & L
    I --> M

1. Modeling the Three-Level Namespace & AI Models

Databricks Unity Catalog organizes assets in a strict hierarchy: catalog $\rightarrow$ schema $\rightarrow$ table/view. The connector extracts four distinct resource tiers:

A. Catalogs (`unitycatalog_catalogs`)

Captures top-level governance containers, storage roots, catalog owners, and business descriptions.

B. Schemas (`unitycatalog_schemas`)

Captures functional namespaces (e.g. finance, marketing, ml_feature_store), properties, and regional metadata.

C. Tables & Views with Rich Column Types (`unitycatalog_tables`)

Tables are transformed into structured markdown documents detailing table types (MANAGED, EXTERNAL, VIEW), underlying data formats (DELTA, PARQUET), and complete column schemas:

# Table: production.finance.settled_transactions
- **Catalog:** production | **Schema:** finance
- **Table Type:** MANAGED (DELTA)
- **Owner:** finance-data-eng
- **Storage Location:** s3://company-lakehouse/finance/settled_transactions

### Columns
| Column Name | Type | Nullable | Comment |
| :--- | :--- | :--- | :--- |
| `transaction_id` | STRING | No | Unique payment transaction identifier |
| `account_id` | STRING | No | Foreign key to core accounts |
| `amount_cents` | BIGINT | No | Settled amount in cents |
| `settled_at` | TIMESTAMP | No | Monotonic settlement timestamp |

D. AI/ML Model Registry (`unitycatalog_models`)

Unity Catalog registers MLflow models directly within schemas. The connector ingests registered model names, current aliases (e.g., champion, challenger), model versions, run artifacts, and tags.

2. Dynamic Scoping & Forget-on-Delete Invariants

Enterprise data lakehouses often host thousands of tables. Ingesting an entire data lake at once can create unnecessary overhead. The connector supports granular scoping:

source = unitycatalog_source(
    catalogs=["production", "gold_analytics"],
    schemas=["finance", "risk_modeling"],
    include_tables=True,
    include_models=True,
)

Automatic Lifecycle Management ("Forget-on-Delete")

When data engineers decommission legacy tables or drop experimental schemas, stale nodes shouldn't persist in company knowledge graphs:

setattr(source, DOCUMENT_SOURCE_ATTR, "unitycatalog")

@dlt.resource(name="unitycatalog_tables", write_disposition="replace")
def tables_resource():
    ...

By tagging DOCUMENT_SOURCE_ATTR = "unitycatalog" and write_disposition="replace", Cognee's internal diff engine detects removed tables upon each resync. Removed tables, columns, and model entries are automatically pruned from FalkorDB/NetworkX and Qdrant, guaranteeing 100% data consistency.

3. End-to-End Implementation

Here is how easily you can connect Databricks Unity Catalog to Cognee to query enterprise governance:

import asyncio
import os
import cognee
from cognee_community_connector_unitycatalog import unitycatalog_source

async def main():
    # 1. Connect to Databricks Workspace or Unity Catalog OSS
    source = unitycatalog_source(
        base_url=os.environ.get("DATABRICKS_HOST", "https://adb-1234.azuredatabricks.net"),
        token=os.environ["DATABRICKS_TOKEN"],
        catalogs=["production"],
        include_tables=True,
        include_models=True,
    )

    # 2. Ingest into Cognee and build Knowledge Graph
    print("Ingesting Unity Catalog lakehouse metadata...")
    await cognee.add(source)

    print("Running Cognify to construct Governance Graph...")
    await cognee.cognify()

    # 3. Ask natural language questions across schemas & models
    query = "What columns exist in the production finance tables, and what is our champion credit model?"
    print(f"\nQuerying Cognee: '{query}'")

    results = await cognee.search(query)
    for res in results:
        print(f"\n[Governance Answer]:\n{res}")

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

Verification & Standards

100% Offline Testing: 18 comprehensive unit tests pass offline using mocked API payloads (tests/test_unitycatalog.py).
Security Guardrails: Pre-ingestion credential masking strips Databricks tokens from all logs and error traces.
Maintainer Alignment: Fully compliant with Cognee’s DCO sign-off guidelines and Hatchling packaging conventions.

Pull Request & Resources

Pull Request: topoteretes/cognee-community#472
Repository: topoteretes/cognee

How is your organization approaching conversational governance across Databricks and Iceberg catalogs? Share your feedback below!

Automating Post-Mortems: Ingesting PagerDuty Incidents & Service Topologies into Cognee GraphRAG

Pranjul Chaurasiya — Sat, 31 Oct 2026 17:13:17 +0000

Writing incident retrospectives (post-mortems) and debugging recurring production outages are two of the most time-consuming responsibilities for Site Reliability Engineers (SREs).

Engineers often spend hours cross-referencing PagerDuty timelines, responder notes, service dependencies, and past incident histories. When teams attempt to solve this with standard LLMs, they quickly run into context window limits or hallucinated timelines because incidents are tightly bound to time, services, and escalation hierarchies.

With GraphRAG (Knowledge Graph Retrieval-Augmented Generation), an AI system can traverse:
Incident $\rightarrow$ AFFECTS $\rightarrow$ Service $\rightarrow$ ESCALATES_TO $\rightarrow$ Escalation Policy $\rightarrow$ RESOLVED_BY $\rightarrow$ Responder

As part of the WeMakeDevs × Cognee Mergetober Hackathon, I built and submitted the PagerDuty data-source connector (cognee-community-connector-pagerduty) for Cognee.

Here is an architectural walkthrough of how we bridged PagerDuty incidents and service topologies into Cognee.

Architectural Workflow

The connector extracts live incident streams and service architecture from PagerDuty's v2 REST API, cleans and redacts the payloads, and routes them into Cognee's document ingestion pipeline:

flowchart TD
    subgraph PagerDutyAPI ["PagerDuty REST API v2"]
        A["/incidents (Urgencies, Statuses, Timelines)"]
        B["/incidents/{id}/notes (Responder Chronology)"]
        C["/services (Service Catalogs & Dependencies)"]
    end

    subgraph ConnectorCore ["Security & Normalization Core"]
        D["Rate-Limit Backoff (HTTP 429 & Retry-After)"]
        E["Secret Redaction (API Tokens, Keys, PATs)"]
        F["Structured Markdown Serializer"]
    end

    subgraph CogneeCore ["Cognee GraphRAG Layer"]
        G["dlt Resource (cognee_document_source='pagerduty')"]
        H["Cognify Pipeline (LLM Relationship Extraction)"]
        I["Service & Incident Graph (FalkorDB / NetworkX)"]
        J["Semantic Memory (Vector Search)"]
    end

    A & B & C --> D --> E --> F --> G --> H --> I & J

1. Dual-Resource Ingestion: Incidents & Service Catalogs

Incidents in isolation lack critical context without knowing the service topology. The connector ingests two synchronized resources:

A. Live & Historical Incidents (`pagerduty_incidents`)

Captures incident IDs, numbers, titles, urgency levels (high, low), lifecycle statuses (triggered, acknowledged, resolved), created and resolved timestamps, assignees, and crucially, responder notes:

# PagerDuty Incident #48291: Payment Gateway Timeout Surge
- **Urgency:** high | **Status:** resolved
- **Service:** Checkout-API
- **Escalation Policy:** Payments-Primary-Tier1
- **Created At:** 2026-10-31T18:20:00Z
- **Resolved At:** 2026-10-31T18:45:00Z
- **Assignees:** Alex Chen, Maria Garcia

### Responder Timeline & Notes
- **2026-10-31T18:25:00Z (Alex Chen):** Identified 504 gateway timeouts originating from payment vendor webhook endpoint.
- **2026-10-31T18:35:00Z (Maria Garcia):** Toggled circuit breaker to secondary gateway; error rates normalized.

B. Service Topologies (`pagerduty_services`)

Ingests registered microservices, descriptions, alert creation configurations, auto-resolve timeout limits, and team ownership.

2. Production Security & Pre-Ingestion Redaction

Incident titles, descriptions, and responder notes frequently contain sensitive diagnostic payloads pasted during live triage—such as database connection strings, bearer tokens, or AWS credentials.

The PagerDuty connector applies strict pre-ingestion regex redaction across all incident fields before emitting records:

import re

PATTERNS = [
    (re.compile(r"Bearer\s+[A-Za-z0-9\-._~+/]+=*", re.IGNORECASE), "Bearer [REDACTED_TOKEN]"),
    (re.compile(r"AKIA[0-9A-Z]{16}"), "[REDACTED_AWS_KEY]"),
    (re.compile(r"sk-[A-Za-z0-9]{20,}"), "[REDACTED_API_KEY]"),
    (re.compile(r"ghp_[A-Za-z0-9]{36}"), "[REDACTED_GITHUB_PAT]"),
]

def redact_secrets(text: str) -> str:
    for pattern, repl in PATTERNS:
        text = pattern.sub(repl, text)
    return text

3. Cognitive Integration with Cognee

By tagging the source with:

setattr(source, DOCUMENT_SOURCE_ATTR, "pagerduty")

Cognee bypasses generic relational flattening and feeds the Markdown summaries into its entity extraction pipeline.

Querying Incident Memory in Python

import asyncio
import os
import cognee
from cognee_community_connector_pagerduty import pagerduty_source

async def main():
    # 1. Initialize PagerDuty connector
    source = pagerduty_source(
        api_key=os.environ["PAGERDUTY_API_KEY"],
        resources=["incidents", "services"],
        urgencies=["high"],
    )

    # 2. Add to Cognee and run Cognify
    print("Ingesting PagerDuty incidents and service topologies...")
    await cognee.add(source)

    print("Building incident knowledge graph...")
    await cognee.cognify()

    # 3. Query past incidents to automate post-mortems
    query = "What caused the high urgency Checkout-API incident and how did Maria resolve it?"
    print(f"\nQuerying: '{query}'")

    results = await cognee.search(query)
    for res in results:
        print(f"\n[Incident Memory]:\n{res}")

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

Verification & Compliance

Complete Test Coverage: Offline test suite mocking PagerDuty REST API endpoints with zero network dependencies (tests/test_source.py, tests/test_client.py, tests/test_redaction.py).
Resilient Rate-Limiting: Exponential backoff with Retry-After capping to avoid thread locks.
DCO Certified: Clean commit history with Developer Certificate of Origin sign-off.

Check Out the Code

Pull Request: topoteretes/cognee-community#429
Repository: topoteretes/cognee

Have questions about integrating incident response into AI workflows? Feel free to connect in the comments!

Incident Intelligence: Connecting Atlassian Opsgenie to Cognee AI Memory & Knowledge Graphs

Pranjul Chaurasiya — Sat, 31 Oct 2026 17:12:27 +0000

When high-severity incidents (P1/P2) trigger in production, engineering teams are inundated with fragmented information: fired alert payloads, escalation hops, on-call rotation schedules, and post-incident Slack threads.

Traditional Retrieval-Augmented Generation (RAG) fails in incident response because incidents are fundamentally relational:

Which on-call engineer was paged when the authentication microservice breached latency thresholds?
Has this specific database failover happened before across our escalation policies?
What services depend on the upstream system currently firing alerts?

To answer these questions, AI systems need more than simple vector similarity over flat text—they need GraphRAG (Knowledge Graph RAG).

As part of the WeMakeDevs × Cognee Mergetober Hackathon, I built and submitted the Atlassian Opsgenie data-source connector (cognee-community-connector-opsgenie) for Cognee.

In this deep dive, we explore how to turn Opsgenie alert streams, on-call schedules, and escalation trees into an active AI knowledge graph while adhering to strict security and lifecycle invariants.

Architecture: Real-Time SRE Memory

The Opsgenie connector bridges operational telemetry into Cognee’s semantic layer:

flowchart TD
    subgraph OpsgenieAPI ["Atlassian Opsgenie API"]
        A["/v2/alerts (P1-P5 Severities)"]
        B["/v2/schedules (On-Call Rotations)"]
        C["/v2/escalations (Routing Rules)"]
    end

    subgraph ConnectorGuard ["Connector Security & Hygiene"]
        D["SSRF Domain Validator (api.opsgenie.com / api.eu.opsgenie.com)"]
        E["Rate-Limit Backoff (Cap Retry-After at 60s)"]
        F["Credential Redaction (GenieKey & Auth Headers)"]
    end

    subgraph Formatting ["Document & Metadata Serialization"]
        G["Alert Markdown Documents + Entity Properties"]
        H["On-Call Schedule & Participant Trees"]
        I["Escalation Policy Nodes"]
    end

    subgraph CogneeEngine ["Cognee GraphRAG Pipeline"]
        J["dlt Source (DOCUMENT_SOURCE_ATTR='opsgenie')"]
        K["Cognify Pipeline (LLM Entity/Relation Extraction)"]
        L["Vector Store (Embeddings)"]
        M["Knowledge Graph (FalkorDB / NetworkX)"]
        N["Orphan Pruning (Forget-on-Delete)"]
    end

    A & B & C --> D --> E --> F
    F --> G & H & I --> J
    J --> K
    K --> L & M
    J --> N

1. Multi-Resource Ingestion: Alerts, On-Call, & Escalations

Opsgenie is not just a stream of alerts; it represents the operational topology of your engineering organization. The connector structures three distinct resources:

A. Alert Telemetry (`opsgenie_alerts`)

Ingests active and historical alerts, extracting priority (P1 through P5), status (open, acknowledged, closed), tags, source integrations, responder notes, and diagnostic details.

# Opsgenie Alert: Database Connection Pool Exhausted (P1 - CRITICAL)
- **Alert ID:** `4a9b...`
- **Status:** open
- **Created At:** 2026-10-31T20:15:00Z
- **Responders:** SRE-Platform-Team
- **Tags:** `database`, `postgres`, `payment-gateway`
- **Description:** Active connections reached 100% capacity on primary read-write cluster.

B. On-Call Rotations (`opsgenie_schedules`)

Ingests schedules, rotation layers (weekly/daily/custom), timezones, and active on-call participants. This allows Cognee to correlate who was on shift during an incident window.

C. Escalation Routing (`opsgenie_escalations`)

Maps out escalation policies, delay intervals, repeat counts, and notification rules across user and team targets.

2. Hardened Production Security & Maintainer Invariants

Building enterprise connectors requires battle-tested engineering. Here are the core patterns implemented to meet Cognee maintainer review standards:

SSRF & Domain Validation

When traversing pagination links or following @odata.nextLink / next URLs, the client explicitly validates hostnames before attaching credentials:

from urllib.parse import urlparse

ALLOWED_HOSTS = {"api.opsgenie.com", "api.eu.opsgenie.com"}

def validate_target_url(url: str) -> str:
    parsed = urlparse(url)
    if parsed.scheme != "https":
        raise ValueError(f"Insecure scheme: {parsed.scheme}. HTTPS required.")
    if parsed.hostname not in ALLOWED_HOSTS:
        raise ValueError(f"SSRF violation: Host {parsed.hostname} is not an authorized Opsgenie endpoint.")
    return url

Sanitized Error Messaging (Zero Leaks in Sentry/Logs)

HTTP error bodies can contain echoed bearer tokens or sensitive system parameters. The connector never dumps raw HTTP bodies into exceptions:

def raise_sanitized_error(response):
    # Safe error message with HTTP status code only
    msg = f"Opsgenie API error: HTTP {response.status_code} ({response.reason_phrase})"
    err = OpsgenieAPIError(msg)
    # Store scrubbed details privately without exposing to standard str(err)
    err.status_code = response.status_code
    raise err

Capped `Retry-After` Handling

Unbounded retry loops can hold process-wide staging locks in distributed environments. If Opsgenie returns an HTTP 429 with Retry-After: 300, sleeping for 5 minutes can cause downstream deadlocks. The connector caps backoffs to a maximum of 60 seconds:

def calculate_retry_delay(response) -> float:
    header = response.headers.get("Retry-After")
    if header and header.isdigit():
        return min(float(header), 60.0)
    return 2.0

3. The "Forget-on-Delete" Invariant in Action

Incident management data is ephemeral: alerts close, schedules rotate, and temporary escalations expire. If an AI memory system retains resolved or deleted alerts forever, retrieval precision degrades.

By configuring:

setattr(source, DOCUMENT_SOURCE_ATTR, "opsgenie")

and tagging resources with:

write_disposition="replace"

Cognee's ingestion engine recognizes when an alert is resolved or deleted upstream. During synchronization, Cognee computes the snapshot delta and triggers automatic orphan cleanup, pruning stale entities and edges from FalkorDB/NetworkX and Qdrant without corrupting existing historical relationships.

Quickstart: SRE Knowledge Graph in 40 Lines of Python

Here is how you can index your active Opsgenie alerts and on-call schedules into Cognee:

import asyncio
import os
import cognee
from cognee_community_connector_opsgenie import opsgenie_source

async def main():
    # 1. Configure the Opsgenie source
    # Filters to open high-priority alerts while syncing active schedules
    source = opsgenie_source(
        api_key=os.environ["OPSGENIE_API_KEY"],
        base_url="https://api.opsgenie.com",
        query="status: open AND priority: P1",
        include_schedules=True,
        include_escalations=True,
    )

    # 2. Add to Cognee memory and build the Knowledge Graph
    print("Ingesting Opsgenie telemetry...")
    await cognee.add(source)

    print("Running Cognify: Extracting SRE entities and building relationships...")
    await cognee.cognify()

    # 3. Query the SRE Memory
    query = "Who is currently on-call for critical P1 database alerts, and what is the escalation path?"
    print(f"\nQuerying Cognee: '{query}'")

    results = await cognee.search(query)
    for result in results:
        print(f"\n[Insight]: {result}")

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

What Happens Behind the Scenes?

During cognee.cognify(), Cognee extracts entity nodes like:

Alert("Database Connection Pool Exhausted")
Team("SRE-Platform-Team")
Person("Jane Doe", role="Primary On-Call")
EscalationRule("Step 1 -> Page Jane Doe; Step 2 (5 min) -> Page Platform Lead")

It establishes graph edges (ASSIGNED_TO, ESCALATES_TO, AFFECTS_SERVICE), enabling the LLM to give grounded, deterministic operational answers.

Verification & Code Quality

The connector is verified with a comprehensive offline test harness:

14/14 Unit Tests Passing: Zero live network calls using respx HTTP mocking (tests/test_opsgenie.py).
CodeQL CLI Security Audit: 100% clean across 174 rules (python-security-and-quality.qls) with 0 CWE findings.
DCO Certified: Verified Developer Certificate of Origin (git commit -s).

Check Out the Code

The Atlassian Opsgenie connector is open for review in the Cognee Community ecosystem:

Pull Request: topoteretes/cognee-community#478
Upstream Discussion: topoteretes/cognee#5682

Building SRE agents or interested in GraphRAG for operations? Let me know your thoughts or feedback in the comments!

Zero Code Execution: Building a Safe Jupyter Notebook Connector for Cognee GraphRAG

Pranjul Chaurasiya — Sat, 31 Oct 2026 17:12:25 +0000

In modern data science organizations, Jupyter Notebooks (.ipynb) serve as the primary medium for exploratory data analysis (EDA), model training, hypothesis validation, and architectural experiments. However, standard enterprise knowledge graphs and LLM retrieval pipelines have historically struggled to ingest notebooks safely and meaningfully:

Security Vulnerabilities (OWASP LLM02): Ingesting external notebooks that contain untrusted Python or shell magics (!curl, os.system) risks accidental code execution if runtime environments (ipykernel, exec(), eval()) are triggered during parsing.
Noise & Bloat: A single uncleaned notebook can contain tens of megabytes of raw Base64 PNG plots, ANSI color codes from failed stack traces, and verbose progress bar outputs.
Stale Knowledge & Orphaned Entities: When a data scientist deletes or refactors a notebook, stale entity nodes and outdated embeddings often remain indefinitely in vector and graph storage.

As part of the WeMakeDevs × Cognee Mergetober Hackathon, I engineered and submitted the official Jupyter Notebook data-source connector (cognee-community-connector-jupyter) for Cognee.

In this article, I will break down how we solved these challenges through zero code execution passive parsing, pre-ingestion secret redaction, and Cognee’s native forget-on-delete graph lifecycle.

Architecture: From Raw `.ipynb` to Knowledge Graph

Here is the high-level ingestion pipeline from disk to Cognee's GraphRAG engine:

flowchart TD
    A[".ipynb File on Disk"] --> B["Passive JSON Parser (nbformat v3/v4)"]
    B --> C["Cell Content Extractor"]

    subgraph Sanitization ["Pre-Ingestion Sanitization"]
        C --> D["Markdown Cells (Structured Prose)"]
        C --> E["Code Cells (Fenced Code Blocks)"]
        C --> F["Outputs (stdout / text/plain)"]
        F --> G["Noise Filter (Strip ANSI, Drop Base64 Images)"]
        D & E & G --> H["Regex Redactor (Bearer, API Keys, PATs)"]
    end

    H --> I["dlt Document Generator (cognee_document_source='jupyter')"]
    I --> J["Cognee Ingestion Engine"]

    subgraph Cognify ["Cognify Graph Processing"]
        J --> K["Entity & Relationship Extraction (LLM)"]
        K --> L["Knowledge Graph (NetworkX / FalkorDB)"]
        K --> M["Vector Store (Qdrant / LanceDB)"]
    end

    J --> N["write_disposition='replace' (Forget-on-Delete Orphan Cleanup)"]

1. Zero Code Execution: Passive JSON AST Parsing

A critical security requirement when building data pipelines for AI memory is never executing untrusted code. Many naive notebook parsers inadvertently import kernel execution utilities or evaluate expressions.

Under the hood, every .ipynb file is simply a structured JSON document conforming to the Jupyter nbformat schema. Our connector treats notebooks strictly as passive data using Python's standard library:

import json
from pathlib import Path

def parse_notebook_statically(file_path: Path) -> dict:
    """Statically parse .ipynb as raw JSON with zero kernel interaction."""
    with open(file_path, "r", encoding="utf-8") as f:
        nb_data = json.load(f)

    # Validate nbformat version
    nbformat_ver = nb_data.get("nbformat", 4)
    cells = nb_data.get("cells", [])
    return {"version": nbformat_ver, "cells": cells}

By decoupling extraction from ipykernel, the pipeline can ingest thousands of notebooks across legacy versions (nbformat v3 and v4) with zero CPU execution risk and zero dependency overhead.

2. Noise Suppression & Output Filtering

Raw notebook outputs are notoriously noisy. To keep LLM context windows efficient and prevent embedding pollution:

Base64 Image Suppression: Base64-encoded PNG/JPEG plots (image/png, image/jpeg) are stripped entirely from outputs.
ANSI Color Stripping: Terminal escape sequences generated by Python tracebacks (\u001b[0;31m) are removed with compiled regex patterns.
Character Caps: Verbose stdout logs are capped to configurable lengths (max_output_chars=2000) with explicit truncation markers.

import re

ANSI_REGEX = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")

def clean_traceback_output(text: str) -> str:
    """Remove ANSI terminal color escape sequences from traceback strings."""
    return ANSI_REGEX.sub("", text)

3. Pre-Ingestion Credential & Secret Redaction

Data scientists frequently forget hardcoded test credentials in scratch cells before saving notebooks. Emitting unredacted API tokens into shared vector databases or external LLM inference endpoints violates enterprise privacy boundaries.

The connector runs a compiled regex redaction pipeline over all Markdown prose, code blocks, and output buffers before records leave the local connector sandbox:

OpenAI Keys: sk-[A-Za-z0-9]{20,}
GitHub PATs: ghp_[A-Za-z0-9]{36}
AWS Access Keys: AKIA[0-9A-Z]{16}
Bearer Tokens: Bearer [A-Za-z0-9\-\._~\+\/]+=*

REDACTION_RULES = [
    (re.compile(r"Bearer\s+[A-Za-z0-9\-._~+/]+=*", re.IGNORECASE), "Bearer [REDACTED_TOKEN]"),
    (re.compile(r"ghp_[A-Za-z0-9]{36}"), "[REDACTED_GITHUB_PAT]"),
    (re.compile(r"sk-[A-Za-z0-9]{20,}"), "[REDACTED_API_KEY]"),
    (re.compile(r"AKIA[0-9A-Z]{16}"), "[REDACTED_AWS_KEY]"),
]

def redact_secrets(content: str) -> str:
    for pattern, replacement in REDACTION_RULES:
        content = pattern.sub(replacement, content)
    return content

4. The Cognee Graph Invariant: "Forget-on-Delete"

In GraphRAG architectures, handling deletions and moves is just as important as handling additions. If a team removes an obsolete experimentation notebook, its graph nodes shouldn't mislead the LLM.

To solve this, the connector integrates directly with Cognee's internal DOCUMENT_SOURCE_ATTR mechanism:

from cognee.tasks.ingestion.dlt_utils import DOCUMENT_SOURCE_ATTR

# 1. Flag the source for document routing
setattr(source, DOCUMENT_SOURCE_ATTR, "jupyter")

# 2. Use full snapshot replacement disposition
@dlt.resource(name="notebooks", write_disposition="replace")
def notebooks_resource():
    for notebook in discover_and_parse_notebooks():
        yield notebook

Why this matters:

Bypasses Relational Flattening: DOCUMENT_SOURCE_ATTR = "jupyter" signals to Cognee that incoming items are rich documents rather than relational database tables, automatically routing them to the Cognify entity-extraction pipeline.
Automated Graph Pruning: Setting write_disposition="replace" allows Cognee's background orphan cleanup job to compute differences between sync snapshots and automatically prune removed notebooks from both vector collections and the FalkorDB/NetworkX knowledge graph.
Content Deduplication: Unchanged notebooks retain identical deterministic data_id content hashes, skipping redundant LLM inference calls on subsequent runs.

In Action: Ingesting Notebooks and Querying the Knowledge Graph

Using the connector in a Cognee pipeline takes just a few lines of asynchronous Python:

import asyncio
import cognee
from cognee_community_connector_jupyter import jupyter_source

async def main():
    # 1. Initialize connector pointing to data science repositories
    source = jupyter_source(
        workspace_dir="./analytics_workspace",
        include_outputs=True,
        max_output_chars=2000,
        redact_secrets=True,
    )

    # 2. Add source to Cognee staging and cognify into knowledge graph
    print("Ingesting notebooks...")
    await cognee.add(source)

    print("Extracting entities and building graph relationships...")
    await cognee.cognify()

    # 3. Query the unified notebook knowledge graph
    query = "What data preprocessing steps and evaluation metrics were used for customer churn?"
    print(f"\nQuerying: '{query}'")
    results = await cognee.search(query)

    for idx, match in enumerate(results, 1):
        print(f"\n[Result {idx}]:\n{match}")

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

When Cognee processes the notebooks, it identifies concepts like RandomForestClassifier, AUC-ROC, and data_leakage_mitigation, creating cross-notebook edges that connect experimental hypotheses with quantitative findings.

Verification & Maintainer Standards

Following the Cognee core maintainer review standards:

100% Offline Testing: 27 unit tests pass offline with zero external network or kernel dependencies (tests/test_source.py, tests/test_parser.py, tests/test_redaction.py).
Dependency Hygiene: Clean separation using modern pyproject.toml configuration with flexible bounds (cognee>=1.6.3,<2).
DCO Certified: Verified Developer Certificate of Origin (git commit -s).

Conclusion & What's Next

By combining passive JSON parsing, pre-ingestion redaction, and Cognee's GraphRAG lifecycle, data science teams can safely turn their institutional notebook repositories into interactive, searchable memory.

You can inspect the connector and open PR under the Cognee community repository:

Pull Request: topoteretes/cognee-community#373
Cognee Core: topoteretes/cognee

Have questions about GraphRAG, Cognee connector architecture, or notebook sanitization? Drop a comment below or connect on GitHub!

DEV Community: Pranjul Chaurasiya

Software Supply Chain Memory: Connecting JFrog Artifactory to Cognee GraphRAG

Architecture: Supply Chain Knowledge Graph

1. Multi-Resource Ingestion: Repositories, Artifacts, & Build Runs

A. Repository Topologies (artifactory_repositories)

B. Cryptographic Artifact Integrity (artifactory_artifacts)

C. CI/CD Build Provenance & SBOM (artifactory_builds)

2. Invariants & Security Standards

Zero Code Execution

Pre-Ingestion Credential Masking

"Forget-on-Delete" via Snapshot Sync

3. Querying Supply Chain Memory in Python

Verification & Code Quality

Pull Request & Resources

Bridging the Lakehouse: Ingesting Databricks Unity Catalog into Cognee Knowledge Graphs

Architectural Blueprint

1. Modeling the Three-Level Namespace & AI Models

A. Catalogs (unitycatalog_catalogs)

B. Schemas (unitycatalog_schemas)

C. Tables & Views with Rich Column Types (unitycatalog_tables)

D. AI/ML Model Registry (unitycatalog_models)

2. Dynamic Scoping & Forget-on-Delete Invariants

Automatic Lifecycle Management ("Forget-on-Delete")

3. End-to-End Implementation

Verification & Standards

Pull Request & Resources

Automating Post-Mortems: Ingesting PagerDuty Incidents & Service Topologies into Cognee GraphRAG

Architectural Workflow

1. Dual-Resource Ingestion: Incidents & Service Catalogs

A. Live & Historical Incidents (pagerduty_incidents)

B. Service Topologies (pagerduty_services)

2. Production Security & Pre-Ingestion Redaction

3. Cognitive Integration with Cognee

Querying Incident Memory in Python

Verification & Compliance

Check Out the Code

Incident Intelligence: Connecting Atlassian Opsgenie to Cognee AI Memory & Knowledge Graphs

Architecture: Real-Time SRE Memory

1. Multi-Resource Ingestion: Alerts, On-Call, & Escalations

A. Alert Telemetry (opsgenie_alerts)

B. On-Call Rotations (opsgenie_schedules)

C. Escalation Routing (opsgenie_escalations)

2. Hardened Production Security & Maintainer Invariants

SSRF & Domain Validation

Sanitized Error Messaging (Zero Leaks in Sentry/Logs)

Capped Retry-After Handling

3. The "Forget-on-Delete" Invariant in Action

Quickstart: SRE Knowledge Graph in 40 Lines of Python

What Happens Behind the Scenes?

Verification & Code Quality

Check Out the Code

Zero Code Execution: Building a Safe Jupyter Notebook Connector for Cognee GraphRAG

Architecture: From Raw .ipynb to Knowledge Graph

1. Zero Code Execution: Passive JSON AST Parsing

2. Noise Suppression & Output Filtering

3. Pre-Ingestion Credential & Secret Redaction

4. The Cognee Graph Invariant: "Forget-on-Delete"

Why this matters:

In Action: Ingesting Notebooks and Querying the Knowledge Graph

Verification & Maintainer Standards

Conclusion & What's Next

A. Repository Topologies (`artifactory_repositories`)

B. Cryptographic Artifact Integrity (`artifactory_artifacts`)

C. CI/CD Build Provenance & SBOM (`artifactory_builds`)

A. Catalogs (`unitycatalog_catalogs`)

B. Schemas (`unitycatalog_schemas`)

C. Tables & Views with Rich Column Types (`unitycatalog_tables`)

D. AI/ML Model Registry (`unitycatalog_models`)

A. Live & Historical Incidents (`pagerduty_incidents`)

B. Service Topologies (`pagerduty_services`)

A. Alert Telemetry (`opsgenie_alerts`)

B. On-Call Rotations (`opsgenie_schedules`)

C. Escalation Routing (`opsgenie_escalations`)

Capped `Retry-After` Handling

Architecture: From Raw `.ipynb` to Knowledge Graph