Every method maps to one lifecycle

Claims flow through five stages. The API is organized around this lifecycle — every method has a clear purpose in how knowledge enters, gets queried, and self-corrects.

1
Ingest
ingest()
ingest_batch()
ingest_text()
ingest_chat()
ingest_slack()
2
Query
query()
explain()
search()
resolve()
path_exists()
3
Understand
schema()
quality_report()
knowledge_health()
find_gaps()
find_bridges()
consensus()
blindspots()
4
Correct
retract()
retract_cascade()
trace_downstream()
at()
impact()
audit()
drift()
5
Track
ingest_inquiry()
check_inquiry_matches()
open_inquiries()
fragile()
stale()
source_reliability()
hypothetical()

Setup

FunctionDescription
attestdb.quickstart(path, vocabs=["bio"], curator="heuristic") Create a database with vocabularies and curator in one call.
attestdb.open(path, backend="kuzu") Open or create a database. Backends: "kuzu", "rust", "auto".
db.configure_curator(model) Set the curator. "heuristic" (offline) or a provider name (see below).
db.register_vocabulary(namespace, vocab) Register entity types, predicates, and constraints for a domain.
db.register_predicate(predicate_type, constraints) Register a single predicate with subject/object type constraints.
db.register_payload_schema(predicate_type, schema) Register a JSON schema for payload validation on a predicate type.
db.close() Close the database. Also works as a context manager: with attestdb.open(...) as db:

Adding Claims

MethodDescription
db.ingest(subject, predicate, object, provenance, confidence=None, payload=None) Add a single claim. Returns the claim ID.
db.ingest_batch(claims) Add many claims at once. Returns BatchResult with .ingested and .duplicates.
db.ingest_chat(messages, conversation_id="", extraction="heuristic") Extract claims from a conversation. Accepts OpenAI/Anthropic message format.
db.ingest_chat_file(path, platform="auto") Extract claims from a file: ChatGPT export ZIP, JSON conversation, or plain text.
db.ingest_slack(path, channels=None, extraction="heuristic") Extract claims from a Slack workspace export ZIP. Optionally filter by channel name.
db.ingest_text(text, source_id="") Extract claims from raw text.
db.curate(claims) Run claims through the curator before ingesting. Returns stored/skipped/flagged.

Extraction modes

ModeAPI Key?When to use
"heuristic"NoExplicit relational text ("X depends on Y"). Fast and free.
"llm"YesNuanced or implicit relationships. Deeper understanding.
"smart"YesLarge volumes. Heuristic first, LLM only for new content. Saves cost.

Querying

MethodDescription
db.query(entity, depth=2) Get a full picture of an entity: relationships, narrative summary, confidence scores.
db.explain(entity, depth=2) Same as query() but also returns timing and candidate counts.
db.search(query, k=10) Semantic search via embeddings. Returns the top-k closest claims.
db.resolve(entity_name) Resolve an entity name to its canonical normalized ID.
db.get_entity(entity_id) Get an entity summary: name, type, claim count.
db.claims_for(entity, predicate_type=None) Get raw claims for an entity. Optionally filter by predicate.
db.claims_by_content_id(content_id) Get all claims about the same fact (corroboration group).
db.list_entities(entity_type=None, min_claims=0) List all entities. Optionally filter by type or minimum claim count.
db.path_exists(entity_a, entity_b, max_depth=3) Check if two entities are connected.
db.find_paths(entity_a, entity_b, max_depth=3, top_k=5) Find the top-k paths between two entities with edge details.
db.raw_query(cypher) Escape hatch: run a raw Cypher query against the underlying Kuzu store.

Understanding Your Knowledge Base

MethodDescription
db.schema() What entity types, predicates, and relationship patterns exist, with counts.
db.stats() Entity count, claim count, index size.
db.quality_report() Health metrics: single-source entities, source distribution, knowledge density.
db.find_gaps(expected_patterns) Find entities that are missing expected relationships.
db.find_bridges(top_k=20) Predict potential connections between currently-unlinked entities.
db.find_confidence_alerts() Find entities with reliability concerns (single-source, stale data).
db.knowledge_health() Quantified health score (0-100) with metrics: multi-source ratio, corroboration, freshness, source diversity, confidence trend.

Health score breakdown

MetricWeightWhat it measures
Multi-source ratio30%Fraction of entities backed by more than one source
Corroboration ratio25%Fraction of claims independently confirmed
Freshness20%Recency of claims (30-day half-life decay)
Source diversity15%Number of distinct source types
Confidence trend10%Whether confidence is improving over time

Knowledge Topology

MethodDescription
db.compute_topology() Run Leiden community detection on the claim graph. Returns a TopologyResult.
db.topics() Get the topic hierarchy from the last topology computation.
db.density_map() Get density metrics per topic: claim count, source diversity, coverage.
db.cross_domain_bridges() Find entities that connect different knowledge domains.
db.query_topic(topic_id) Get all entities belonging to a specific topic.
db.generate_structural_embeddings() Generate SVD-based graph embeddings for all entities.
db.generate_weighted_structural_embeddings() Generate confidence-weighted SVD graph embeddings.

Provenance and Trust

MethodDescription
db.retract(source_id, reason) Mark all claims from a source as retracted.
db.retract_cascade(source_id, reason) Retract a source and mark anything that depended on it as degraded.
db.trace_downstream(claim_id) See what other claims depend on a specific claim.
db.at(timestamp) Query the knowledge base as it was at a specific point in time.

Knowledge Intelligence

Nine methods that answer questions only a provenance-tracking database can answer.

MethodDescription
db.impact(source_id) If this source is retracted, how many claims and entities are affected? Returns direct count, downstream count, and affected entity list.
db.blindspots(min_claims=5) Find entities backed by only a single source, knowledge gaps, and low-confidence areas.
db.consensus(topic) How many independent sources agree about an entity? Returns agreement ratio, claims by source, and corroborated content IDs.
db.fragile(max_sources=1) Find claims backed by few independent sources — these are your weakest knowledge.
db.stale(days=90) Find claims that haven't been corroborated or updated within the given time period.
db.audit(claim_id) Full provenance chain: who said it, what corroborates it, how many claims depend on it.
db.drift(days=30) How has knowledge changed? New claims, new entities, retracted sources, confidence trends.
db.source_reliability(source_id=None) Per-source corroboration and retraction rates. Pass a source_id for one, or omit for all sources.
db.hypothetical(claim) What-if analysis: would this claim corroborate existing knowledge? Does it fill a gap?

Research Questions

MethodDescription
db.ingest_inquiry(question, subject, object, predicate_hint="") Register a question you want answered.
db.open_inquiries() List all unanswered questions.
db.check_inquiry_matches(subject_id=None, object_id=None) Check if new claims match any open questions.

Backup and Restore

MethodDescription
db.snapshot(dest_path) Copy the database to a backup directory.
AttestDB.restore(src_path, dest_path) Restore a database from a snapshot.

LLM Providers

Set the environment variable for your provider, then configure:

ProviderEnvironment VariableConfigure
GroqGROQ_API_KEYdb.configure_curator("groq")
OpenAIOPENAI_API_KEYdb.configure_curator("openai")
AnthropicANTHROPIC_API_KEYdb.configure_curator("anthropic")
DeepSeekDEEPSEEK_API_KEYdb.configure_curator("deepseek")
GrokGROK_API_KEYdb.configure_curator("grok")
OpenRouterOPENROUTER_API_KEYdb.configure_curator("openrouter")
GLMGLM_API_KEYdb.configure_curator("glm")

No API key? Use "heuristic" mode — it works entirely offline.

Data Types

Adding a claim

db.ingest(
    subject=("name", "type"),          # e.g. ("api-gateway", "service")
    predicate=("relationship", "class"), # e.g. ("depends_on", "depends_on")
    object=("name", "type"),             # e.g. ("redis", "service")
    provenance={
        "source_type": "...",           # What kind of source
        "source_id": "...",             # Identifies the specific source
    },
    confidence=0.9,                    # 0.0 to 1.0 (optional)
    payload={...},                     # Any structured data (optional)
)

Query result

frame = db.query("redis")
frame.focal_entity           # The entity you queried
frame.claim_count            # Number of claims about it
frame.direct_relationships   # Connected entities with predicates and confidence
frame.narrative              # Human-readable summary

Batch input

from attestdb import ClaimInput

claims = [
    ClaimInput(
        subject=("api-gateway", "service"),
        predicate=("depends_on", "depends_on"),
        object=("redis", "service"),
        provenance={"source_type": "config_management", "source_id": "k8s"},
    ),
    # ... more claims
]
result = db.ingest_batch(claims)

Event Hooks

Subscribe to lifecycle events. Callbacks run synchronously after the operation completes. Errors in callbacks are logged, never propagated — your pipeline keeps running.

MethodDescription
db.on(event, callback) Register a callback for a lifecycle event.
db.off(event, callback) Remove a registered callback.

Events

EventKwargsFires when
"claim_ingested"claim_id, claim_inputAfter each ingest() call
"claim_corroborated"content_id, countA newly ingested claim matches an existing one
"source_retracted"source_id, reason, claim_idsAfter retract()
"inquiry_matched"inquiry_id, claim_idA newly ingested claim answers an open inquiry

Example

def on_new_claim(claim_id, claim_input, **kw):
    print(f"New claim: {claim_id}")

def on_corroboration(content_id, count, **kw):
    print(f"Corroborated! {count} independent sources")

db.on("claim_ingested", on_new_claim)
db.on("claim_corroborated", on_corroboration)

Agent Integration

Two ways for external agents to read and write Attest — choose based on your agent framework.

MCP Server (Model Context Protocol)

For Claude Desktop, Claude Code, and any MCP-compatible agent. Ships as a CLI tool.

$ pip install attestdb[mcp]
$ ATTEST_DB_PATH=my.db attest-mcp

Exposes 15 tools (ingest_claim, query_entity, search_entities, knowledge_health, retract_source, etc.) and 2 resources (attest://entities, attest://schema) over stdio transport.

Claude Desktop configuration

{
  "mcpServers": {
    "attest": {
      "command": "attest-mcp",
      "env": {
        "ATTEST_DB_PATH": "my_knowledge.db"
      }
    }
  }
}

REST API

For web-based agents, custom integrations, or any HTTP client. Mounted at /api/v1/ in the Attest Console.

$ pip install attest-console
$ attest-console my.db
MethodPathDescription
POST/api/v1/claimsIngest a single claim
POST/api/v1/claims/batchBulk-ingest claims
POST/api/v1/claims/textExtract claims from text
GET/api/v1/entitiesList entities
GET/api/v1/entities/{id}Get entity summary
GET/api/v1/entities/{id}/claimsClaims about an entity
GET/api/v1/entities/{id}/contextFull context frame
GET/api/v1/paths/{a}/{b}Find paths between entities
POST/api/v1/retractRetract a source
GET/api/v1/schemaSchema descriptor
GET/api/v1/statsDatabase statistics
GET/api/v1/healthKnowledge health metrics
GET/api/v1/qualityQuality report
GET/api/v1/insights/bridgesBridge predictions
GET/api/v1/insights/gapsConfidence alerts

Example

# Ingest a claim via REST
curl -X POST http://localhost:8877/api/v1/claims \
  -H "Content-Type: application/json" \
  -d '{
    "subject": ["api-gateway", "service"],
    "predicate": ["depends_on", "dependency"],
    "object": ["redis", "service"],
    "source_type": "k8s_manifest",
    "source_id": "deploy/prod"
  }'

# Query an entity
curl http://localhost:8877/api/v1/entities/redis/context

# Check knowledge health
curl http://localhost:8877/api/v1/health