Direct Integration

Python SDK

Import Mithril directly in your Python code. No HTTP server needed — the same governance pipeline used by the MCP server and REST API is available as a simple async class.

Quick Start

example.py

import asyncio

from mithril import Mithril


async def main():
    firewall = Mithril()
    await firewall.setup()

    # Submit a memory claim through the trust gate
    result = await firewall.remember(
        text="Passwords must be hashed using Argon2id",
        source="Security Policy",
        author="policy_admin",
    )

    print(result.status)                       # AdmissionStatus.WARN or ACCEPT
    print(result.trust_breakdown.final_score)  # normalized 0-1 trust score
    print(result.decision_reason)              # human-readable gate decision
    print(result.trust_breakdown.reasons)      # explainable breakdown list

    # Query verified memory only
    answer = await firewall.recall("How should we hash passwords?")
    print(answer)

    # Query with metadata (candidates count, blocked count)
    recall = await firewall.recall_with_metadata("How should we hash passwords?")
    print(recall.answer)
    print(recall.candidate_count)
    print(recall.blocked_count)


asyncio.run(main())

Installation

Clone and install

terminal

git clone <repo-url>
cd hack-ideas2
uv pip install -e ".[dev]"
# or: pip install -e ".[dev]"

Configure environment

Copy .env.example to .env and fill in your LLM API key (needed for contradiction detection):

terminal

# macOS / Linux / WSL
cp .env.example .env

# Windows PowerShell
Copy-Item .env.example .env

# Set LLM_API_KEY, LLM_ENDPOINT, and LLM_MODEL in .env.
# The optional MITHRIL_LLM_MODEL overrides LLM_MODEL for Mithril's own LLM calls.

Mithril Class Reference

asyncsetup()

Initialize SQLite stores (audit, quarantine, reputation). Must be called once before using the firewall.

firewall = Mithril()
await firewall.setup()

asyncremember(text, source, author, metadata)

Submit a memory claim through the full governance pipeline: credential redaction → source reputation lookup → contradiction detection → trust scoring → admission gate → storage or quarantine → audit logging → reputation update.

Parameter	Type	Required	Description
`text`	str	required	The claim or fact to evaluate.
`source`	str	required	Channel or origin — e.g. "Slack", "Security Policy".
`author`	str	optional	Who produced the claim. Defaults to "unknown".
`metadata`	dict \| None	optional	Additional metadata to attach to the claim.

Returns an AdmissionResult with the full decision and provenance.

asyncrecall(query) → str

Query only verified memory. Returns the answer as a string. Poisoned / quarantined data is excluded. Secrets are re-scrubbed on output.

asyncrecall_with_metadata(query) → RecallResult

Same as recall() but returns a RecallResult with answer, candidate_count, and blocked_count.

asyncget_audit_trail() → list[dict]

Returns the full audit log — every claim evaluated, with status, score, source, reason, and timestamp.

asyncget_quarantine() → list[dict]

Returns all quarantined / rejected / review claims with provenance.

asyncget_reputation() → list[dict]

Returns all sources with current live reputation, prior, delta, and accept/block counts.

asyncget_stats() → FirewallStats

Aggregate metrics: total evaluated, accepted, warned, reviewed, quarantined, rejected, block rate, and average trust score.

asyncreset()

Wipe all Cognee memory and all SQLite stores (audit, quarantine, reputation). Reseeds reputation priors. Used for clean demos.

asyncimprove()

Run Cognee graph enrichment on the verified memory dataset. Builds richer connections between accepted claims.

Key Data Models

📦

AdmissionResult

.claim — MemoryClaim

.trust_breakdown — TrustScoreBreakdown

.status — AdmissionStatus enum

.decision_reason — human-readable string

.cognee_dataset — set if stored in Cognee

.redacted_secrets — list of secret kinds found

📊

TrustScoreBreakdown

.final_score — normalized 0–1

.source_reputation — live reputation used

.contradiction_penalty — 0–1 LLM score

.corroboration_bonus — 0–0.3

.freshness_bonus — 0–0.05

.content_danger_penalty — 0–1 LLM score

.reasons — explainable breakdown list

🚦

AdmissionStatus

ACCEPT — stored in Cognee (verified)

WARN — stored but flagged

REVIEW — held for human review

QUARANTINE — isolated in SQLite

REJECT — rejected outright

🔍

RecallResult

.query — original question

.answer — verified-only answer

.candidate_count — verified chunks found

.blocked_count — quarantined claims excluded

Ingestion Connectors

Bulk-ingest real message feeds through the governance pipeline:

slack_ingest_example.py

from mithril import (
    Mithril,
    ingest_slack_export,
    load_generic_file,
    ingest_messages,
)

firewall = Mithril()
await firewall.setup()

# Ingest a Slack export JSON
results = await ingest_slack_export(firewall, "data/slack_export.json")
print(f"{len(results)} messages processed")

# Ingest a generic .txt or .jsonl file
messages = load_generic_file("data/claims.jsonl")
results = await ingest_messages(
    firewall, messages, source="Internal Wiki"
)

# Each result is an AdmissionResult with full provenance
for r in results:
    print(f"{r.status.value}: {r.claim.text[:50]}")

💬

Slack Export

Parses the standard workspace-export JSON (array of message objects or object with messages key). Bot/system noise is automatically skipped. Author and timestamp are preserved as real provenance.

📄

Generic File

One claim per line (.txt) or one JSON object per line (.jsonl with text, source,author keys). Lines that fail to parse are skipped with a warning.

Standalone Utilities

utilities.py

from mithril import (
    compute_trust_score,
    redact_secrets,
    contains_secret,
    scan_secrets,
    ReputationStore,
)

# Check a string for secrets
has_secret = contains_secret("my-api-key: sk-proj-abc123xyz")
print(has_secret)  # True

# Redact secrets and get match details
redacted, matches = redact_secrets(
    "Use token ghp_abc123def456ghi789jkl012mno345pqr678"
)
print(redacted)   # "Use token [REDACTED:github_token]"
print(matches[0].kind)  # "github_token"

# Scan without redacting
matches = scan_secrets("Bearer eyJhbGciOiJIUzI1NiIs...")
for m in matches:
    print(f"{m.kind}: chars {m.start}–{m.end}")