pip install amfs amfs-adapter-http

import os

from amfs import AgentMemory
from amfs_adapter_http import HttpAdapter

mem = AgentMemory(
    agent_id="test",
    adapter=HttpAdapter(
        base_url="https://amfs-login.sense-lab.ai",
        api_key=os.environ["AMFS_API_KEY"],
    ),
)
print(mem.stats().total_entries)   # what your account already knows

import os

from amfs import AgentMemory, OutcomeType
from amfs_adapter_http import HttpAdapter

mem = AgentMemory(
    agent_id="deploy-agent",
    adapter=HttpAdapter(
        base_url="https://amfs-login.sense-lab.ai",
        api_key=os.environ["AMFS_API_KEY"],
    ),
)

# 1. Get briefed: compiled, evidence-tagged context for what you are about to touch.
for digest in mem.briefing(entity_path="acme/checkout", compact=True):
    print(digest.summary.get("narrative"))

# 2. Recall specifics by meaning. No key, no exact path.
hits = mem.retrieve("why does the checkout worker restart loop", entity_path="acme/checkout")
for hit in hits:
    print(hit.entry.key, round(hit.score, 2), hit.entry.evidence_status)

# ... do the work, recording what you did ...
mem.record_action("deploy_rollback", {"service": "checkout", "to_version": "v41"})

# A remembered fix did not work? Say so before trying the next thing.
mem.record_attempt(summary="restarted worker per runbook; queue still stuck")

# 3. Write what a future agent would need.
mem.write(
    "acme/checkout", "risk-worker-restart-loop",
    "Worker restarts loop when the Redis URL has no db index; add /0.",
    memory_type="belief", confidence=0.8,
)

# 4. Commit the outcome. This is where learning happens.
mem.commit_outcome(
    "INC-2291", OutcomeType.SUCCESS,
    task_input="checkout worker restart looping in prod",
    response_text="Added the db index to REDIS_URL, worker stable for 30 min.",
)

import os

from amfs import AgentMemory
from amfs_adapter_http import HttpAdapter

mem = AgentMemory(
    agent_id="deploy-agent",
    adapter=HttpAdapter(
        base_url="https://amfs-login.sense-lab.ai",
        api_key=os.environ["AMFS_API_KEY"],
    ),
)

from amfs import AgentMemory

mem = AgentMemory(agent_id="my-agent", adapter=adapter)

from amfs_core.importance import ImportanceEvaluator

class MyEvaluator(ImportanceEvaluator):
    def evaluate(self, entity_path, key, value):
        score = 0.8 if "critical" in str(value).lower() else 0.3
        return score, {"criticality": score}

mem = AgentMemory(agent_id="my-agent", adapter=adapter, importance_evaluator=MyEvaluator())

with AgentMemory(agent_id="my-agent", adapter=adapter) as mem:
    mem.write("svc", "key", "value")
    entry = mem.read("svc", "key")
# Watchers, TTL sweepers and background threads are cleaned up

server_mem = AgentMemory(agent_id="server", adapter=adapter)
as_bob = server_mem.as_agent("bob-agent")   # same adapter, Bob's identity, its own read log

from amfs import MemoryType

entry = mem.write(
    "checkout-service",              # entity_path — what this knowledge is about
    "retry-pattern",                 # key
    {"max_retries": 3},              # value — any JSON-serializable data
    confidence=0.85,                 # optional, default 1.0
    pattern_refs=["retry"],          # optional cross-references to other keys
    memory_type=MemoryType.FACT,     # fact (default), belief, experience or procedure
    shared=True,                     # False = private to this agent
)
print(entry.version)   # 1 — writing the same key again creates version 2

from datetime import datetime, timedelta, timezone

mem.write(
    "svc", "temp-flag", {"active": True},
    ttl_at=datetime.now(timezone.utc) + timedelta(hours=24),
)

entry = mem.read("checkout-service", "retry-pattern")

if entry:
    print(entry.value, entry.version, entry.confidence, entry.evidence_status)

# With a confidence floor
entry = mem.read("svc", "pattern", min_confidence=0.5)

# From a specific branch
entry = mem.read("svc", "pattern", branch="experiment-1")

# What do *I* know about this? Only this agent's own versions, private ones included.
mine = mem.recall("checkout-service", "retry-pattern")

# Everything this agent has written, optionally for one entity.
brain = mem.my_entries("checkout-service")

# Read a specific key from a specific agent. Tracked as a cross-agent read.
theirs = mem.read_from("deploy-agent", "checkout-service", "deploy-config")

# Who have I learned from?
mem.cross_agent_reads()    # {'deploy-agent': [{'entity_path': ..., 'key': ..., 'read_count': 2}]}
mem.agents_i_read_from()   # ['deploy-agent']

entries = mem.list()                                     # everything visible to this agent
entries = mem.list("checkout-service")                   # one entity
entries = mem.list("checkout-service", include_superseded=True)
entries = mem.list("checkout-service", branch="experiment-1")

results = mem.search(
    query="retry strategy",              # full-text query
    entity_path="checkout-service",      # or entity_paths=["svc-a", "svc-b"] to merge several
    min_confidence=0.5,
    max_confidence=1.0,
    agent_id="review-agent",             # filter by author
    since=datetime(2026, 3, 1, tzinfo=timezone.utc),   # filter by time (timezone-aware)
    pattern_ref="retry-pattern",         # filter by pattern reference
    sort_by="confidence",                # "confidence" | "recency" | "version" | "priority"
    limit=100,
    depth=3,                             # tier depth (1=hot, 2=hot+warm, 3=all)
)

hot_only = mem.search(query="retry strategy", depth=1)   # hot tier
hot_warm = mem.search(query="retry strategy", depth=2)   # hot + warm
all_tiers = mem.search(query="retry strategy")            # all (default)

from amfs import RecallConfig

scored = mem.search(
    query="how do we handle retries?",
    recall_config=RecallConfig(semantic_weight=0.5, recency_weight=0.3, confidence_weight=0.2),
)
for item in scored:
    print(f"{item.entry.key} — score={item.score:.3f}")

hits = mem.retrieve("checkout worker restart loop", entity_path="acme/checkout", limit=5)
for hit in hits:
    print(hit.entry.entity_path, hit.entry.key, round(hit.score, 3), hit.entry.evidence_status)

hits = mem.retrieve(
    "customer did not receive the invoice email",
    entity_path="acme/support",
    include_priors=True,
    candidate_actions=["resolve:resend_email", "resolve:update_payment_method", "resolve:refund"],
    situation="invoice-not-received",     # optional stable label for this kind of task
    compact=True,
)
priors = mem.last_priors
print(priors["recommendation"]["mode"])              # "act" | "explore" | "escalate" | "abstain"
print(priors["recommendation"].get("suggested_action"))
print(priors["action_priors"])                        # per action: won / lost here on similar tasks

from datetime import datetime, timedelta, timezone

versions = mem.history("checkout-service", "retry-pattern")
for v in versions:
    print(f"v{v.version} — confidence: {v.confidence} — {v.provenance.written_at}")

# Versions from the last 7 days
since = datetime.now(timezone.utc) - timedelta(days=7)
recent = mem.history("checkout-service", "retry-pattern", since=since)

stats = mem.stats()
print(stats.total_entries, stats.total_entities, stats.total_agents)
print(stats.confidence_avg, stats.outcome_linked_count)

for event in mem.timeline(limit=50):
    print(f"[{event.event_type}] {event.summary}")

writes = mem.timeline(event_type="WRITE", limit=20)
recent = mem.timeline(since=datetime.now(timezone.utc) - timedelta(hours=24))

from amfs import MemoryType

# Facts (default) — stable, objective knowledge
mem.write("svc", "config", {"pool_size": 10}, memory_type=MemoryType.FACT)

# Beliefs — hypotheses; decay faster. Keep confidence below 0.9 and say why.
mem.write("svc", "hypothesis", "Likely an N+1 query, because latency scales with cart size",
          memory_type=MemoryType.BELIEF, confidence=0.6)

# Experiences — records of actions taken; decay slower
mem.write("svc", "action-log", "Added index on user_id", memory_type=MemoryType.EXPERIENCE)

# Procedures — how to do a task: ordered steps, preconditions, what to do on failure.
# Decay slowest; the briefing lists them in their own section.
mem.write("svc", "procedure-rollback", {
    "goal": "roll back a bad checkout deploy",
    "steps": [{"text": "scale down canary", "action_key": "k8s:scale"},
              {"text": "promote previous image", "action_key": "deploy:promote"}],
    "preconditions": {"runtime": "python3.12"},
}, memory_type=MemoryType.PROCEDURE)

# External inputs that informed the decision — in the causal chain, not in storage.
mem.record_context("pagerduty-incidents", "3 SEV-1 in the last 24h for checkout", source="PagerDuty API")

# Consequential actions — deploys, rollbacks, edits, refunds. Record failures too.
mem.record_action("deploy_rollback", {"service": "checkout", "to_version": "v41"},
                  result='{"status": "rolled_back"}', duration_ms=1430)
mem.record_action("shell", {"cmd": "pip install -r requirements.txt"},
                  result=stderr, success=False, action_key="shell:pip_install")

mem.record_attempt(summary="restarted worker per runbook; queue still stuck")
mem.record_attempt(outcome_type=OutcomeType.FAILURE,
                   causal_entry_keys=["acme/support/fix-restart"])

from amfs import OutcomeType

updated = mem.commit_outcome(
    "INC-1042", OutcomeType.SUCCESS,
    task_input="checkout worker restart looping in prod",     # the request, as it arrived
    response_text="Added the db index to REDIS_URL; stable.",  # your final answer, in full
    verified_by="ci",                                          # who decided it, if not the agent
    evidence={"run_id": "98765"},
    attributes={"customer": "acme", "task_type": "deploy"},   # dimensions to filter runs by later
)
for e in updated:
    print(e.entity_path, e.key, e.confidence, e.evidence_status)

chain = mem.explain()
print(chain["session_id"])
print(chain["causal_entries"])     # entries read, with the version seen at read time
print(chain["external_contexts"])  # record_context() inputs

# Filter by outcome reference
chain = mem.explain(outcome_ref="INC-1042")

for t in mem.list_traces(entity_path="acme/checkout", limit=10):
    print(t.id, t.agent_id, t.outcome_ref, t.outcome_type, t.created_at)

trace = mem.get_trace(trace_id)
print(trace.task_input, trace.response_text)
for entry in trace.causal_entries:
    print(f"  read {entry.entity_path}/{entry.key} v{entry.version}")
for call in trace.tool_calls:
    print(f"  did {call.tool_name} {call.arguments} success={call.success}")

mem.set_session_attributes({"customer": "acme", "task_type": "deploy"})

resp = client.chat.completions.create(...)
mem.record_llm_call(
    resp.model, resp.usage.prompt_tokens, resp.usage.completion_tokens,
    provider="openai", cost_usd=0.00033, latency_ms=412.0,
)

mem.commit_outcome("deploy-42", OutcomeType.SUCCESS, task_input=..., response_text=...)

briefs = mem.briefing(entity_path="myapp/checkout-service", limit=5)

for digest in briefs:
    print(digest.digest_type)   # "entity", "agent_brief", "source", or "connection_map"
    print(digest.scope)         # the entity path, agent ID, or source ID
    print(digest.summary)       # structured summary (varies by digest type)
    print(digest.entry_count)   # number of source entries compiled
    print(digest.compiled_at)   # when the digest was last compiled

from amfs import AgentMemory, Run

mem = AgentMemory(agent_id="ci-bot", adapter=adapter)
run = Run(mem)

# 1. Before the agent acts: guidance scoped to this run.
guidance = run.begin(
    "fix the failing CI on PR 42",
    entity_path="acme/ci",
    model="gpt-4o", agent_version="ci-bot@1.4.0", runtime="python3.12",
    candidate_actions=["shell:pip_install", "edit:requirements", "shell:pytest"],
)
if guidance.should_inject():          # strength is "strong" or "thin"
    system_prompt = guidance.text + "\n\n" + system_prompt

# 2. After each consequential tool call: record it; on failure, ask for a hint.
hint = run.on_tool_result(
    "shell", {"cmd": "pip install -r requirements.txt"},
    result=stderr, success=False, action_key="shell:pip_install",
)
if hint is not None and hint.should_inject():
    messages.append({"role": "system", "content": hint.text})

# When the approach you were following did not work and you switch, name
# the entry you followed (the keys are in guidance.text) or the top hit:
run.attempt_failed("pinning urllib3 did not resolve the conflict",
                   causal_entry_keys=cited or [guidance.top_key])

# 3. When the run ends: the outcome, who decided it, and what it acted on.
run.complete(True, verified_by="ci", evidence={"run_id": "98765"},
             response_text=final_answer,
             causal_entry_keys=cited or [guidance.top_key])

# 4. What this run learned, as a claim the record can follow.
run.learn("pip install fails on a urllib3/requests conflict", "edit:requirements", True,
          "pinning urllib3<2 resolves it; re-running the install does not")

mem = AgentMemory(agent_id="ci-bot", adapter=adapter, branch="experiment-1")
mem.checkout("main")                                            # back to main
entry = mem.read("acme/ci", "retry-pattern", branch="experiment-1")   # one call, another branch

checkout = mem.scope("checkout-service")

checkout.write("retry-pattern", {"max_retries": 5}, confidence=0.9)
entry = checkout.read("retry-pattern")
all_entries = checkout.list()
results = checkout.search(min_confidence=0.5)
history = checkout.history("retry-pattern")
scope_info = checkout.info()

# Readonly scopes prevent accidental writes
readonly = mem.scope("checkout-service", readonly=True)
readonly.write("key", "value")  # raises PermissionError

scopes = mem.list_scopes()         # every entity path that contains entries
info = mem.info("checkout-service")
# ScopeInfo(path, entry_count, avg_confidence, keys, oldest, newest)

print(mem.tree())
# myapp (5)
#   auth (2)
#   checkout-service (3)

with mem.transaction("update retry configs") as tx:
    tx.write("checkout-service", "retry-pattern", {"max_retries": 5})
    tx.write("checkout-service", "timeout-config", {"timeout_ms": 3000})
    tx.write("payment-service", "retry-pattern", {"max_retries": 2})
    print(tx.pending_count)  # 3
# All writes committed atomically on exit

print(tx.commit.id)
print(len(tx.entries))

commits = mem.commit_log(limit=10)
for c in commits:
    print(f"{c.id} — {c.message} ({len(c.entries)} entries)")

commit = mem.get_commit("abc123...")
ancestor = mem.common_ancestor(commit_a_id, commit_b_id)

diff = mem.diff("checkout-service", "retry-pattern")                 # two most recent versions
diff = mem.diff("checkout-service", "retry-pattern", old_version=1)  # against a specific version

patch = mem.create_patch("checkout-service", "retry-pattern")
patch = mem.create_patch("checkout-service", "retry-pattern", source_version=2)

result = mem.verify("checkout-service")   # or mem.verify() for everything

print(result["total_checked"])
print(result["valid"])
print(result["corrupted"])     # list of corrupted entry info
print(result["chain_breaks"])  # list of chain break info

edges = mem.graph_neighbors(
    "checkout-service/retry-pattern",
    direction="both",
    depth=2,
    min_confidence=0.5,
    limit=50,
)
for edge in edges:
    print(f"{edge.source_entity} --{edge.relation}--> {edge.target_entity}")

mem.set_profile(
    description="Handles deployment decisions for checkout-service",
    default_branch="main",
    auto_context_paths=["checkout-service", "payment-service"],
    tags=["deploy", "infra"],
)

mem.declare_capability(
    "deployment-analysis",
    description="Analyzes deployment readiness based on CI and recent changes",
    entity_paths=["checkout-service", "payment-service"],
)

mem.set_contracts([
    {
        "entity_path": "checkout-service",
        "key_pattern": "retry-*",
        "min_confidence": 0.5,
        "max_confidence": 1.0,
        "required_fields": ["max_retries"],
        "ttl_required": False,
        "description": "Retry configuration entries",
    }
])

agents = mem.discover_agents(capability="deployment-analysis")
agents = mem.discover_agents(entity_path="checkout-service")

from amfs import ConflictPolicy

# Raise StaleWriteError if another agent wrote since our last read
mem = AgentMemory(agent_id="my-agent", adapter=adapter, conflict_policy=ConflictPolicy.RAISE)

# Or merge
def merge(our_last_read, current, new_value):
    return {**current.value, **new_value}

mem = AgentMemory(agent_id="my-agent", adapter=adapter, on_conflict=merge)

rooms = mem.my_rooms()                 # rooms available to this agent
mem.room_join("room-123")
info = mem.room_info("room-123")
updates = mem.room_updates("room-123", since="2026-05-01T00:00:00Z")

mem.room_discuss(
    "room-123",
    "I've updated the retry config based on recent incidents",
    message_type="update",
    addressed_to="deploy-agent",
)
messages = mem.room_discussions("room-123", limit=20)

mem.room_leave("room-123")             # snapshots are preserved

session = mem.negotiate_create(
    "room-123", "Retry Config Agreement",
    description="Agree on shared retry parameters", max_rounds=5,
)

mem.negotiate_propose(
    "room-123", session["id"],
    action="propose",
    proposal={"max_retries": 3, "backoff_ms": 1000},
    rationale="Based on P99 latency data from last week",
)

mem.negotiate_respond("room-123", session["id"], action="accept",
                      rationale="Aligns with our SLA requirements")

status = mem.negotiate_status("room-123", session["id"])

from amfs import (
    # Classes
    AgentMemory,
    MemoryScope,
    Run,
    Guidance,

    # Models
    MemoryEntry,
    MemoryStats,
    MemoryType,
    OutcomeType,
    OutcomeRecord,
    Provenance,
    ProvenanceTier,
    DecisionTrace,
    Event,
    RecallConfig,
    ScoredEntry,
    SearchQuery,
    SemanticQuery,
    SessionMetadata,
    ConflictPolicy,
    QualityIssue,
    QualityReport,

    # Abstract base
    EmbedderABC,
)