AgentKeeper

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.

Why this exists

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.

Install

pip install agentkeeper-ai

Zero 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]'         # Everything

Five things AgentKeeper does that nothing else does

1. Identity that survives everything

Principles 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.

import agentkeeper

agent = 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.

2. Same memory, switch models

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.

3. TTL for GDPR — memory that expires itself

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 protected

4. Graph traversal — structured relations alongside prose memory

Facts 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}

5. Plug into Claude Desktop via MCP

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 anthropic

claude_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.

6. Checkpoints — snapshot, restore, survive a crash

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 gone
agent = agentkeeper.load("aria").restore(snap.snapshot_id)
# identity, facts, graph, and execution_state are back

agentkeeper.diff(snap_a, snap_b)   # factual diff: facts added/removed/modified
agent.list_checkpoints()           # every snapshot, oldest first

Reconstruction 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.py

7. Own your AI memory — import ChatGPT, Claude, and Gemini

Every 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.

import agentkeeper
from agentkeeper.importers import (
    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.

Architecture

       ┌──────────────────────────────────────────────────────────────┐
       │                  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.

Full API tour

Memory primitives

agent.fact("budget: 50k EUR", importance=0.9)          # stable semantic fact
agent.event("contract signed", when="2026-05-15")      # episodic, time-anchored
agent.principle("never share PII")                     # protected, survives all compression
agent.remember("favourite colour: blue")               # tier inferred automatically

Semantic recall

results = agent.recall("money allocated to the project", top_k=5)
for fact, score in results:
    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.

Cognitive compression

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.

Graph relations

agent.link("Acme", "owns", "Globex", confidence=1.0)
agent.link("Acme", "signed_contract", "ThinkLanceAI", ttl="2y")

# BFS traversal
related = agent.find_related("Acme", max_hops=2, direction="out")

# Introspect triples
for triple in agent.triples:
    print(triple)  # Triple('Acme' -[owns]-> 'Globex', conf=1.00)

TTL on facts and triples

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 items

Encrypted storage

from agentkeeper.storage.encrypted_sqlite import EncryptedSQLiteStorage

key = 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.

Async API

import asyncio, agentkeeper

async def main():
    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 = await asyncio.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.

LangChain integration

from agentkeeper.integrations.langchain import LangChainCognitiveProvider
from langchain_core.prompts import ChatPromptTemplate

provider = LangChainCognitiveProvider(agent, model="gpt-4o")
prompt = ChatPromptTemplate.from_messages([
    ("system", provider(message="What's our budget?")),
    ("human", "{input}"),
])

Custom provider profile

from agentkeeper import CognitiveProfile, PromptFormat, register_profile

register_profile(CognitiveProfile(
    provider="my-llm",
    format=PromptFormat.SECTIONS,
    effective_context_tokens=10_000,
))

GDPR / compliance

export = agent.gdpr_export()   # all facts, triples, identity — JSON-serialisable
agent.purge_expired()          # remove anything whose TTL has elapsed
agent.forget(fact_id)          # remove a single fact by id
agentkeeper.delete("aria")     # permanently remove an agent from storage

Observability

snapshot = 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, ...}
# }

Production checklist

Type-safe — py.typed shipped, mypy-strict compatible.
Typed exceptions — AgentKeeperError root, subclasses for every failure mode.
Structured logging — namespaced under agentkeeper.*, opt-in, NullHandler default.
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.

Configuration

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`

Try it in 30 seconds — no API key needed

pip install agentkeeper-ai
python examples/demo.py

The 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.

Roadmap

v1.1 ✅ — TTL, graph layer, encrypted storage, MCP server, LangChain + CrewAI integrations, persistent sqlite-vec index, 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.

Contributing

Issues, ideas, and PRs welcome. See CONTRIBUTING.md.

License

MIT. See LICENSE.

Built by

ThinkLanceAI — Tom Anciaux Berner — cognitive infrastructure for AI systems.

tom@thinklanceai.com

Name		Name	Last commit message	Last commit date
Latest commit History 52 Commits
.github/workflows		.github/workflows
agentkeeper		agentkeeper
benchmark		benchmark
docs		docs
examples		examples
marketing		marketing
tests		tests
.env.example		.env.example
.gitignore		.gitignore
CHANGELOG.md		CHANGELOG.md
CONTRIBUTING.md		CONTRIBUTING.md
LICENSE		LICENSE
README.md		README.md
pyproject.toml		pyproject.toml
requirements.txt		requirements.txt

Folders and files

Latest commit

History

Repository files navigation

AgentKeeper

Why this exists

Install

Five things AgentKeeper does that nothing else does

1. Identity that survives everything

2. Same memory, switch models

3. TTL for GDPR — memory that expires itself

4. Graph traversal — structured relations alongside prose memory

5. Plug into Claude Desktop via MCP

6. Checkpoints — snapshot, restore, survive a crash

7. Own your AI memory — import ChatGPT, Claude, and Gemini

Architecture

Full API tour

Memory primitives

Semantic recall

Cognitive compression

Graph relations

TTL on facts and triples

Encrypted storage

Async API

LangChain integration

Custom provider profile

GDPR / compliance

Observability

Production checklist

Configuration

Try it in 30 seconds — no API key needed

Roadmap

Contributing

License

Built by

About

Topics

Resources

License

Contributing

Uh oh!

Stars

Watchers

Forks

Releases 2

Packages 0

Uh oh!

Contributors

Uh oh!

Languages

Packages