DEV Community: Siddham Jain

How I connected Apollo's sales data to cognee so an AI agent can answer questions about it

Siddham Jain — Fri, 30 Oct 2026 18:15:07 +0000

Sales teams keep track of the people and companies they sell to in a CRM, which is a database built for customer records. Apollo.io is one of these tools. It also runs sequences, which are scheduled series of outreach messages sent to a contact.

An AI model can't answer questions about any of that on its own, because every conversation with it starts from zero. cognee is an open source project that gives AI agents a memory. You feed it your data, it builds a knowledge graph (a map of the things in your data and how they connect), and an agent can search that map later.

A connector is the piece of code that moves data from a tool like Apollo into cognee. I built the Apollo connector for Mergetober, a month of open source work on cognee integrations, and it was merged on October 19.

What the connector does

Once your Apollo data is in cognee, you can ask something like "Which contacts at Acme are in an active sequence?" and get an answer based on your own records.

Copying the data once would be simple. The harder part is that a CRM changes every day, so the connector has to handle two things on every later run:

Only process records that changed. Without this, each run would send your whole CRM through the AI processing step again.
Remove records you deleted in Apollo. If you delete a contact, your agent shouldn't keep answering questions about them.

Apollo makes both of these harder than you'd expect, and most of this post is about why.

Trying it

You'll need an Apollo API key, which means a paid plan or a trial, because free plans don't have API access to contact and account search. You'll also need an API key for an AI model provider, since cognee uses one to build the graph.

The connector is in the cognee-community repo, and you run it from there with uv:

git clone https://github.com/topoteretes/cognee-community.git
cd cognee-community/packages/connector/apollo
uv sync
export APOLLO_API_KEY="your-apollo-key"

Put your model key in a .env file as LLM_API_KEY, then run the example:

uv run python examples/example.py

The part of the example that does the work looks like this:

source = apollo_source(api_key=api_key, resource_name="apollo_crm_main")

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

dataset_name is the place in cognee where this data lives. resource_name is a name you pick for this Apollo account, and it has to stay the same between runs. write_disposition="merge" tells cognee to update existing records. Leave it out and the second sync wipes the first one.

The connector reads your contacts, your company accounts and your sequences. It leaves emails, phone numbers and social profiles out on purpose, and it doesn't use any of the Apollo endpoints that cost credits.

Problem 1: Apollo doesn't tell you what changed

Many services stamp each record with the time it was last updated, so a connector can ask for everything newer than its last run. Apollo has an updated_at field, but it doesn't work for this. I enrolled a contact in a sequence on a live account and the contact's timestamp stayed the same, even though that's the kind of change you'd want your agent to know about. Company records don't come back with the timestamp at all.

So the connector reads every record on each run and turns it into text. It then computes a fingerprint of that text, a short code that changes whenever the text changes. It saves the fingerprints, and on the next run any record with the same fingerprint as before is skipped.

The downside is that reading everything uses up requests, and Apollo limits how many you can make per minute, per hour and per day. The connector stops before it hits a limit, saves its place, and picks up from there on the next run. A large account on a small plan will take several runs to get through.

Problem 2: Apollo doesn't tell you what was deleted

Apollo has no list of deleted records. All the connector can see is that a record it knew about isn't in the list anymore, and that doesn't prove it was deleted. The run might have stopped early, or the list might have been cut off, because Apollo search stops at 50,000 records.

So the connector only acts on a list it read all the way to the end. For each record missing from that list, it asks Apollo about that one record directly. It forgets the record only if Apollo says it's gone, or if the record no longer matches the filters you chose.

What the code review found

Before the connector was merged, goran-radonic, a cognee maintainer, reviewed it over two rounds and left 31 inline comments. Two of the problems they found show how easily the delete logic could go wrong.

The first was about two Apollo accounts sharing one dataset. Both used the same default name, so in the reviewer's test, syncing the second account removed the first account's contacts, and the next sync of the first account removed the second's. The name is now required, and a run refuses to continue from saved progress that belongs to a different Apollo team.

The second was in how I recognized a deleted record. Apollo reports one with an error message, and my check was too loose: a different error, such as "This contact does not exist in the stage you filtered", would also have counted as deleted. The check now accepts only the exact messages Apollo uses for deleted records, plus a plain 404.

Did it work?

I tested it on a real Apollo trial account with records I made for the purpose: two companies, four contacts and one sequence.

What I did	What happened
First sync	7 documents stored
Synced again with no changes	all 7 skipped
Edited one contact's job title	6 skipped, 1 updated
Deleted a contact and a company in Apollo	both removed, nothing else touched

That test covered getting data in and taking it out, and it ran on the version I first submitted. After the review changes I checked the fixes with the automated tests, which run against a fake version of Apollo, and with a few specific calls to the real API. The merged version has 83 tests.

Limits

Accounts over 50,000 records need stage or list filters, otherwise deleted records aren't detected.
Each contact keeps its latest 50 activity events.
Apollo plans to remove the error fields the delete check reads on 16 February 2027, and it hasn't documented a replacement yet, so that check will need updating before then.
The dataset contains names, job titles and companies, so share it only with people who are allowed to see your CRM.

The code and setup details are in the Apollo connector folder.

Imagine your ai agent fixing its own fu*ckups, here's how the rewind actually works

Siddham Jain — Sun, 26 Jul 2026 14:14:54 +0000

When an AI agent fails in production, you get told that it failed. You almost never get to
answer the only question that matters: what if the third agent had made a different call?
You can't replay the run. You can't isolate the step that broke it. You can't test a fix without
re-running the whole thing and hoping.

I spent a hackathon building Rewind — a "time machine" for multi-agent runs — to see if
SigNoz's traces plus LangGraph's checkpointer could actually answer that question. They can. This
post is the honest version: how the replay works, and the three things that nearly broke it.

The setup: a crew that ships a subtly wrong fix

The observed workload is a four-agent coding crew (Planner → Coder → Tester → Reviewer) that
fixes a GitHub-style issue. The issue is deliberately ambiguous:

"An order line for more than 10 units should receive a 10% discount."

The tests, however, require the discount at exactly 10 units. A weak model reads the spec
literally and writes:

if quantity > 10:            # what the weak model wrote
    subtotal *= 0.90

and the boundary test fails with a very real assertion:

def test_bulk_discount_boundary():
    assert cart_total([(10.0, 10)]) == 90.0
E   assert 100.0 == 90.0

That's the failure I want to reproduce and fix from a trace, not by re-reading logs.

Making every agent step a span

I instrument the crew two ways. openinference-instrumentation-langchain auto-captures the
LangChain/LangGraph internals, and a tiny custom context manager wraps each node so the span
carries exactly the attributes I'll want to query later:

@contextmanager
def node_span(config, name, inputs):
    with tracer.start_as_current_span(f"crew.{name}") as span:
        span.set_attribute("rewind.node", name)
        span.set_attribute("rewind.thread_id", thread_id(config))
        span.set_attribute("rewind.input", short(inputs))
        yield span

def record_llm(span, model, resp):
    usage = resp.usage_metadata or {}
    span.set_attribute("gen_ai.request.model", model)
    span.set_attribute("gen_ai.usage.input_tokens", usage.get("input_tokens", 0))
    span.set_attribute("gen_ai.usage.output_tokens", usage.get("output_tokens", 0))
    span.set_attribute("rewind.cost_usd", round(cost, 6))

I used OpenTelemetry's GenAI semantic conventions
(gen_ai.request.model, gen_ai.usage.*_tokens) for the model/token fields so they're not
bespoke, and a rewind. namespace for the app-specific bits. Spans export over OTLP/HTTP to
SigNoz's ingester on :4318. The rewind.thread_id is the thread that stitches an entire run
together — remember it, it's the whole trick.

Here's what lands in SigNoz — every planner/coder/tester/reviewer step, queryable:

Gotcha #1: getting SigNoz up when Docker Desktop is gone

I deployed SigNoz with Foundry (casting.yaml → casting.yaml.lock → cast), which is a
genuinely clean one-command install. But my machine had no Docker Desktop, and the leftovers
fought me: docker pointed at a dangling symlink, docker compose was a broken cli-plugin, and
Foundry's cast died on docker-credential-desktop executable not found. The fixes, in order:

brew install colima docker and colima start — a lightweight Docker engine, no Desktop.
relink the compose plugin into ~/.docker/cli-plugins/.
delete the credsStore: "desktop" line from ~/.docker/config.json so Foundry could authenticate to the registry.

None of this is in a tutorial. If you're on a clean Mac without Docker Desktop, Colima + fixing
config.json is the path.

Reading the telemetry back out of SigNoz

Here's the part I care about most: Rewind doesn't keep its own trace store. To render a run's
timeline, the backend queries SigNoz for its own spans via the v5 query API:

POST /api/v5/query_range
{
  "requestType": "raw",
  "compositeQuery": {"queries": [{"type": "builder_query", "spec": {
    "signal": "traces",
    "filter": {"expression":
      "service.name = 'rewind-crew' AND rewind.thread_id = 'run-64c55532' AND rewind.node EXISTS"},
    "selectFields": [
      {"name": "rewind.node",  "fieldContext": "attribute"},
      {"name": "rewind.cost_usd", "fieldContext": "attribute"},
      {"name": "gen_ai.usage.input_tokens", "fieldContext": "attribute"}
    ]
  }}]}
}

The filter.expression string syntax is the thing to get right in v5 — it's a mini query
language (AND, =, EXISTS), and fieldContext: "attribute" vs "resource" matters for
whether SigNoz finds your field. Once that clicked, the same query grouped by rewind.node
became a reliability heatmap across every run: pass rate, average cost per node, and where
failures surface. Across 24 seeded runs mine sat at a 25% pass rate with the failures localizing
to the Tester and Reviewer nodes — which is exactly right, because that's where a wrong > vs
>= gets caught.

Gotcha #2: the replay is real because LangGraph makes it real

The reproduce-and-fork isn't a simulation, and that's the honest core of the project. The crew
writes to a LangGraph SqliteSaver checkpointer, and the backend shares that same file. So to
fork a run I use LangGraph's native time-travel:

history = list(graph.get_state_history(config))     # every checkpoint of the run
snap = pick_snapshot_before(history, node="coder")  # rewind to just before the coder
graph.update_state(snap.config, values=overrides)   # change model / guardrail
graph.invoke(None, fork_config)                      # re-run everything downstream, live

The nodes after the checkpoint actually re-execute with the new state, and the forked run emits
its own spans — so it becomes a second trace in SigNoz, not a diffed copy. In Rewind's UI you
can see the reproduced timeline, the exact code the model wrote, and the real pytest output that
failed:

10` endraw , the failing pytest, and the fork panel" width="800" height="556">

When I fork the failing run with a spec guardrail and a stronger model, the coder rewrites the
line as if quantity >= 10, and the same pytest that was red goes green. Failed and fixed, side
by side, both backed by real traces — and the fix costs a few more tokens, which the cost deltas
make honest.

Gotcha #3: know when not to depend on the live query

SigNoz v0.134 moved its auth around (login is under /api/v2/sessions, not where older guides
say), and I burned time before making a call I'd defend: the checkpointer is the source of
truth for the timeline; the SigNoz query is enrichment. If the query path is configured, the
timeline gets real tokens/cost from SigNoz and every card deep-links to the live trace; if it
isn't, the app still reproduces and forks from the checkpointer. A demo that hard-depends on one
network call to a freshly-booted observability stack is a demo that dies on stage. Degrade
gracefully.

The same telemetry drives a dashboard and a failure-rate alert, both built programmatically via
the SigNoz API — the alert plots the failure count against a threshold, straight from the Query
Builder:

What I'd tell my past self

Pick your span attributes before you write a single query. I renamed fields twice because I hadn't decided that rewind.thread_id was the join key. Design the schema first.
fieldContext is not optional trivia. In the v5 query API, marking a field attribute vs resource is the difference between data and an empty result set.
Dashboard widgets need a layout. I posted widgets with no id/layout and got a dashboard that rendered the empty "Welcome" state — the panels were stored but never placed on the grid.
Let the framework do the replay. I almost hand-rolled a "re-run from step N" engine. LangGraph's get_state_history + update_state + invoke(None, cfg) already is that engine, and it's the reason the fork is trustworthy instead of a mock.
Two proofs beat one. pytest is the hard gate (it can't be argued with); an LLM-as-judge score rides along as a soft quality signal on the timeline. Different jobs, both useful.

Conclusion

Rewind reproduces a production agent failure, forks one decision, and proves the fix with a real
passing test — and none of it required a custom trace store or a faked replay. SigNoz's spans
already held the full I/O and cost; LangGraph's checkpointer already held the executable state.
The work was joining them on one thread id. If you're building agents, that pairing —
OpenTelemetry traces for what happened, a checkpointer for do it again differently — is worth
stealing.

Code: github.com/siddham-jain/rewind. Built for the
Agents of SigNoz hackathon.