Cognitive continuity infrastructure for long-lived AI agents.
Your agent survives model switches, crashes, context-window limits, and restarts — with the same identity, memory, and priorities it had before.
Agents don't fail because they forget facts. They fail because they lose cognitive continuity — their state, priorities, and identity drift the moment the model changes, the context window fills, or the process restarts.
AgentKeeper treats this as a systems problem, not a memory problem.
pip install agentkeeper-aiZero required dependencies. No external infrastructure. Storage defaults to local SQLite.
pip install 'agentkeeper-ai[anthropic]'# Claude
pip install 'agentkeeper-ai[openai]'# GPT + OpenAI embeddings
pip install 'agentkeeper-ai[gemini]'# Gemini
pip install 'agentkeeper-ai[semantic]'# Local embeddings (sentence-transformers)
pip install 'agentkeeper-ai[mcp]'# MCP server (Claude Desktop, Cursor, Codex)
pip install 'agentkeeper-ai[encrypted]'# Encrypted storage at rest
pip install 'agentkeeper-ai[all]'# EverythingPrinciples and constraints are protected — exempt from every compression pass, injected into every reconstructed context, regardless of token budget. They survive decay, consolidation, contradiction arbitration, model switches, and process restarts.
importagentkeeperagent=agentkeeper.create(agent_id="aria", provider="anthropic")
agent.set_identity(
name="Aria",
role="EU insurance broker copilot",
principles=["never share PII without explicit consent"],
constraints=["EU data residency only"],
)
agent.principle("always confirm budget changes in writing")
agent.fact("client: Acme Corporation", importance=0.95)
agent.event("contract signed", when="2026-05-15")
agent.save()
# 100 compression cycles later — identity intact.The cognitive state is reconstructed in the format each model expects. XML for Claude, labelled sections for GPT-4, narrative prose for Gemini, terse tokens for Ollama. One agent, four runtimes, zero rewrites.
agent=agentkeeper.load("aria", provider="anthropic")
response=agent.ask("What do we know about Acme?")
agent.switch_provider("openai").save()
response=agent.ask("Same question, different model.")
# Memory and identity are intact. Format has changed. Nothing broke.Facts and graph triples accept a TTL. When it lapses, purge_expired() removes them. No manual cleanup. Compliant by default.
agent.fact("session token: abc123", ttl="1h")
agent.fact("audit log reference: Q1-2026", ttl="90d")
agent.link("Acme", "signed_contract", "ThinkLanceAI", ttl="2y")
agent.purge_expired() # removes what's lapsed, keeps what's protectedFacts are prose. Triples are structure. Both live in the same agent, with their own retention and TTL policy.
agent.link("Acme", "owns", "Globex")
agent.link("Globex", "located_in", "BE")
agent.link("Alice", "works_at", "Acme", confidence=0.9)
related=agent.find_related("Acme", max_hops=2, direction="out")
# {"Globex": 1, "BE": 2, "Alice": 1}AgentKeeper ships an MCP server. Any MCP-aware client — Claude Desktop, Cursor, Claude Code — gets full access to the agent's cognitive layer without writing a line of integration code.
agentkeeper-mcp --agent-id aria --provider anthropicclaude_desktop_config.json:
{
"mcpServers": {
"aria": {
"command": "agentkeeper-mcp",
"args": ["--agent-id", "aria", "--provider", "anthropic"]
}
}
}Available tools over MCP: add_fact, recall, set_identity, link, find_related, compress, health, gdpr_export, purge_expired, checkpoint, restore, list_checkpoints.
Freeze the full cognitive state into an immutable, content-hashed
snapshot. Restore it after a crash, a context-window overflow, a model
switch, or a process restart. Attach an opaque execution_state payload
(current file, pending task, todos) — AgentKeeper stores and returns it
verbatim; it never interprets or runs it.
snap=agent.checkpoint(
label="before refactor",
execution_state={
"current_file": "auth.py",
"pending_task": "finish RS256 migration",
},
)
# crash, restart, switch model — the process memory is goneagent=agentkeeper.load("aria").restore(snap.snapshot_id)
# identity, facts, graph, and execution_state are backagentkeeper.diff(snap_a, snap_b) # factual diff: facts added/removed/modifiedagent.list_checkpoints() # every snapshot, oldest firstReconstruction is deterministic — the same snapshot always rebuilds the same cognitive state, verified by a content hash. Behaviour is not guaranteed: AgentKeeper restores the agent's state, not the model's next decision. The crash-recovery demo runs with no API key:
python examples/crash_recovery.pyEvery AI platform remembers things about you. None of them show you what, and none of them talk to each other. AgentKeeper imports the official data exports of all three into one scoped, queryable memory — then tells you exactly what each platform knows that the others don't.
importagentkeeperfromagentkeeper.importersimport (
ChatGPTImporter, ClaudeImporter, GeminiImporter,
compare_scopes, scope_inventory,
)
agent=agentkeeper.create("my-memory")
ChatGPTImporter("chatgpt-export.zip").apply(agent)
ClaudeImporter("claude-export.zip").apply(agent)
GeminiImporter("takeout.zip").apply(agent)
print(scope_inventory(agent))
# {"platform:chatgpt": {"EVENT": 142, "PREFERENCE": 9, ...},# "platform:claude": {"EVENT": 87, "FACT": 4, ...}, ...}result=compare_scopes(agent, "platform:chatgpt", "platform:claude")
# ChatGPT only: ['Building a touring platform for electronic music']# Claude only: ['Working on an EU AI Act observatory']# Shared: [('My name is Tom, developer in Brussels', 1.0)]Every imported fact is namespaced (metadata["scope"] == "platform:chatgpt"), so knowledge from one platform never leaks into
another's context unless you promote it. Matching is deterministic —
normalized content equality plus a difflib ratio, no embeddings, no
LLM calls, reproducible to the byte. PII (account email, full name)
stays out unless you pass include_pii=True. Archives are parsed in
memory with zip-bomb and path-traversal guards; nothing is ever
extracted to disk.
Also ships parse_memory_summary(text) for the prompt-generated memory
summaries used by the 2026 cross-platform migration flows.
┌──────────────────────────────────────────────────────────────┐
│ AgentKeeper Public API │
│ agent.remember() · agent.recall() · agent.ask() │
│ agent.compress() · agent.link() · agent.find_related() │
│ agent.set_identity() · agent.purge_expired() · agent.save() │
└────────────────────────────┬─────────────────────────────────┘
│
┌────────────────────────────▼─────────────────────────────────┐
│ Cognitive Reconstruction Engine (CRE) │
│ Identity injection · importance ranking · semantic boost │
│ Token budget · profile-driven rendering │
└─┬───────────┬────────────┬────────────┬──────────────────────┘
│ │ │ │
┌─────▼─────┐ ┌──▼─────────┐ ┌▼──────────┐ ┌▼──────────────┐
│ Memory │ │ Semantic │ │ Cognitive │ │ Cross-Model │
│ Hierarchy │ │ Recall │ │ Compress │ │ Translation │
│ │ │ │ │ │ │ │
│ working │ │ embeddings │ │ decay │ │ XML (Claude) │
│ episodic │ │ vector idx │ │ consol. │ │ sections (GPT)│
│ semantic │ │ sqlite-vec │ │ contradic │ │ narrative (G.)│
│ archival │ │ │ │ │ │ minimal (Oll.)│
└───────────┘ └────────────┘ └───────────┘ └───────────────┘
│ │
┌─────▼────────────────────────────────────────────▼────────────┐
│ Graph Layer Storage (pluggable) │
│ Triple, TTL, BFS SQLite · Encrypted SQLite · Postgres* │
│ agent.link() AGENTKEEPER_DB, AGENTKEEPER_ENC_KEY │
└───────────────────────────────────────────────────────────────┘
│
┌─────────────────────▼───────────────────┐
│ MCP Server · LangChain · CrewAI │
│ agentkeeper-mcp · langchain_system_prompt│
└──────────────────────────────────────────┘
*Postgres stub available; full implementation in v1.2.
agent.fact("budget: 50k EUR", importance=0.9) # stable semantic factagent.event("contract signed", when="2026-05-15") # episodic, time-anchoredagent.principle("never share PII") # protected, survives all compressionagent.remember("favourite colour: blue") # tier inferred automaticallyresults=agent.recall("money allocated to the project", top_k=5)
forfact, scoreinresults:
print(f"{score:.2f}{fact.content}")Pluggable backends: local sentence-transformers (default, free, offline), OpenAI, or your own. Persistent index via sqlite-vec — survives process restarts without rebuild.
report=agent.compress()
# CompressionReport(# decayed_facts=12,# consolidation={'clusters_found': 3, 'facts_removed': 7},# contradictions={'pairs_found': 2, 'resolutions': 2},# facts_before=120, facts_after=102,# )Three independent passes: decay (exponential half-life on unused facts), consolidation (embedding-based clustering, optional LLM synthesiser), contradiction arbitration (key-value divergence + polarity detection, deterministic winner). Protected facts are immortal.
agent.link("Acme", "owns", "Globex", confidence=1.0)
agent.link("Acme", "signed_contract", "ThinkLanceAI", ttl="2y")
# BFS traversalrelated=agent.find_related("Acme", max_hops=2, direction="out")
# Introspect triplesfortripleinagent.triples:
print(triple) # Triple('Acme' -[owns]-> 'Globex', conf=1.00)agent.fact("session token: abc123", ttl="1h")
agent.fact("temp context: negotiation", ttl="30d")
agent.link("Project Phoenix", "uses_provider", "Anthropic", ttl="P90D")
agent.purge_expired() # returns count of removed itemsfromagentkeeper.storage.encrypted_sqliteimportEncryptedSQLiteStoragekey=EncryptedSQLiteStorage.generate_key()
storage=EncryptedSQLiteStorage(encryption_key=key)
# or via env: AGENTKEEPER_ENCRYPTION_KEY=...AES-128-CBC + HMAC-SHA256 at rest. Progressive migration — plain SQLite rows are upgraded on next save.
importasyncio, agentkeeperasyncdefmain():
agent=agentkeeper.create_async(agent_id="aria", provider="anthropic")
agent.set_identity(name="Aria", role="copilot")
agent.fact("budget: 50k EUR", importance=0.95)
answers=awaitasyncio.gather(
agent.ask("status?", provider="anthropic"),
agent.ask("status?", provider="openai"),
)
asyncio.run(main())Sync and async agents share the same storage. Save with one, load with the other.
fromagentkeeper.integrations.langchainimportLangChainCognitiveProviderfromlangchain_core.promptsimportChatPromptTemplateprovider=LangChainCognitiveProvider(agent, model="gpt-4o")
prompt=ChatPromptTemplate.from_messages([
("system", provider(message="What's our budget?")),
("human", "{input}"),
])fromagentkeeperimportCognitiveProfile, PromptFormat, register_profileregister_profile(CognitiveProfile(
provider="my-llm",
format=PromptFormat.SECTIONS,
effective_context_tokens=10_000,
))export=agent.gdpr_export() # all facts, triples, identity — JSON-serialisableagent.purge_expired() # remove anything whose TTL has elapsedagent.forget(fact_id) # remove a single fact by idagentkeeper.delete("aria") # permanently remove an agent from storagesnapshot=agent.health()
# {# "total_facts": 42,# "critical_facts": 3,# "protected_facts": 5,# "contradicted_facts": 1,# "stale_facts": 4,# "importance_stats": {"mean": 0.71, "min": 0.10, "max": 0.95},# "tier_distribution": {"semantic": 28, "episodic": 10, ...},# "graph": {"total_triples": 8, "protected_triples": 2},# "identity": {"name": "Aria", "principles_count": 3, ...}# }- Type-safe —
py.typedshipped, mypy-strict compatible. - Typed exceptions —
AgentKeeperErrorroot, subclasses for every failure mode. - Structured logging — namespaced under
agentkeeper.*, opt-in,NullHandlerdefault. - Retries — exponential backoff + jitter via
with_retry/with_async_retry. - Tested — 491 tests, CI on Python 3.10 / 3.11 / 3.12.
- Zero breaking changes from v0.1 —
agent.remember(content, critical=True)still works.
| Variable | Default | Purpose |
|---|---|---|
OPENAI_API_KEY | — | OpenAI provider |
ANTHROPIC_API_KEY | — | Anthropic provider |
GEMINI_API_KEY | — | Gemini provider |
OLLAMA_HOST | http://localhost:11434 | Ollama server |
AGENTKEEPER_DB | agentkeeper.db | SQLite path |
AGENTKEEPER_ENCRYPTION_KEY | — | Fernet key for encrypted storage |
AGENTKEEPER_EMBEDDING_PROVIDER | auto | sentence-transformers › openai › mock |
pip install agentkeeper-ai
python examples/demo.pyThe demo runs entirely on provider="mock" and AGENTKEEPER_EMBEDDING_PROVIDER=mock. No keys. No network. Shows identity hardening, compression, cross-model translation, and async — in one file.
- v1.1 ✅ — TTL, graph layer, encrypted storage, MCP server, LangChain + CrewAI integrations, persistent
sqlite-vecindex, GDPR export, health snapshot, cognitive checkpoints (snapshot / restore / diff), 491 tests. - v1.2 — Postgres storage backend (full implementation), TypeScript SDK.
- v1.3 — AgentKeeper Cloud (managed sync). OSS stays feature-complete.
Issues, ideas, and PRs welcome. See CONTRIBUTING.md.
MIT. See LICENSE.
ThinkLanceAI — Tom Anciaux Berner — cognitive infrastructure for AI systems.
tom@thinklanceai.com