SDK methods

Three integration layers — from zero-config closed-loop to full low-level control.

MuBit SDKs expose three layers:

Start with learn or helpers. Drop to raw domains only when you need exact payload control.

Learn module (closed-loop)

Before each LLM call, the learn module retrieves relevant lessons from MuBit and injects them into the system message. After the call, the interaction is ingested automatically. On run end, reflection extracts new lessons.

import mubit.learn
 
mubit.learn.init(api_key="mbt_...", agent_id="my-agent")
 
# All OpenAI/Anthropic/LiteLLM calls now auto-inject lessons + auto-ingest.
 
@mubit.learn.run(agent_id="planner", auto_reflect=True)
def plan_task(task):
    return openai.OpenAI().chat.completions.create(
        model="gpt-4o", messages=[{"role": "user", "content": task}],
    ).choices[0].message.content

Helper method bundles

The learning loop

1. remember()            → ingest facts, traces, lessons
2. record_step_outcome() → per-step process reward signals (optional, for dense RL)
3. reflect()             → LLM extracts lessons from evidence
                            (recurring lessons promote: run → session → global)
4. get_context()         → retrieve relevant lessons for the next call
5. record_outcome()      → reinforce what worked at the run level

With learn, steps 1–3 happen automatically. With helpers, you orchestrate them yourself.

Current helper catalog

Step-level outcomes

Record per-step process rewards for dense RL signal within a run. Use after each agentic step, then reflect with the step outcomes folded in.

client.record_step_outcome(
    step_id="tool_call_1",
    step_name="search_api",
    outcome="success",
    signal=0.8,
    rationale="Found the correct document on first try",
    directive_hint="Keep using search before browsing",
)

The Python reflect() helper only accepts session_id, include_linked_runs, user_id, and last_n_items — there is no include_step_outcomes kwarg. To fold step outcomes into reflection, set the include_step_outcomes field on the wire Reflect request (gRPC/HTTP), or call it from JS:

// Later: reflect with step outcomes included
await client.control.reflect({ run_id: "my-run", include_step_outcomes: true });

Lane-scoped memory

Lanes partition memory within a shared run so each agent sees only relevant entries.

# Ingest into a specific lane
client.remember(content="Planning output: task A depends on B", intent="fact", lane="planning")
 
# Query only the planning lane
result = client.recall(query="task dependencies", lane="planning")
 
# Register an agent with lane participation
client.register_agent(agent_id="planner", role="planner", shared_memory_lanes=["planning", "shared"])

Step-wise reflection

Scope reflection to recent evidence for incremental lesson extraction.

# Reflect over only the 5 most recent items
client.reflect(last_n_items=5)

To reflect with step outcomes included, use the wire include_step_outcomes field (gRPC/HTTP) or the JS control helper:

await client.control.reflect({ run_id: "my-run", include_step_outcomes: true });

Auto-extraction in learn

Enable heuristic extraction of rules, lessons, and facts from LLM responses without an extra LLM call.

mubit.learn.init(
    api_key="mbt_...",
    agent_id="my-agent",
    auto_extract=True,          # extract structured items from LLM responses
    extraction_mode="heuristic", # no LLM call needed
)

When to use what

When to use raw control methods directly

Use client.* when you need one of these explicitly:

• control.ingest plus get_ingest_job job polling
• control.batch_insert
• exact raw request/response debugging against HTTP or gRPC
• advanced or compatibility state-management routes
• control.get_ingest_job, control.get_run_ingest_stats for job polling
• control.list_run_history, control.link_run, control.unlink_run, control.delete_run for run management
• control.context_snapshot / control.contextSnapshot for full context snapshots

Ingest job tracking

remember() and the raw ingest() are asynchronous — they return a job_id and return immediately. If the next step in your code reads the same run (for example, recall() or get_context()), poll get_ingest_job(job_id) until done=True to guarantee the new item is visible. Framework adapters that advertise "synchronous put" semantics (e.g. MubitStore in mubit-langgraph) do this polling for you; the raw SDK does not.

Retrieve per-run statistics with get_run_ingest_stats() when you want aggregate counts without scanning activity.

# Submit + poll until the job is durable
job_resp = client.ingest({"run_id": "my-run", "items": [{"text": "…", "intent": "fact"}]})
while True:
    status = client.get_ingest_job(job_id=job_resp["job_id"])
    if status.get("done"):
        break
 
# Per-run aggregate stats
stats = client.get_run_ingest_stats(run_id="my-run")

Run management

List run history, link, unlink, and delete runs. The server clamps the limit on run-history listing to a max of 1000 (default 100).

# List recent run history (advanced domain)
runs = client.advanced.list_run_history(limit=100)
 
# Link a child run to a parent
client.link_run(run_id="parent-run", linked_run_id="child-run")
 
# Unlink
client.unlink_run(run_id="parent-run", linked_run_id="child-run")
 
# Delete a run and its data
client.delete_run(run_id="old-run")

Context snapshot

Retrieve a full context snapshot for a run, including working memory, attention state, and active goals.

snapshot = client.advanced.context_snapshot(run_id="my-run")

Temporal and quality features

Occurrence time

MuBit tracks two time dimensions for every memory entry: ingestion time (when the system learned it) and occurrence time (when the event actually happened). Set occurrence_time to record when an event happened, separate from when it was ingested.

import time
 
# Event happened 3 days ago, ingested now
client.ingest({
    "run_id": "my-run",
    "items": [{
        "text": "New CI/CD pipeline reduced deployment time by 60%.",
        "intent": "fact",
        "occurrence_time": int(time.time()) - 86400 * 3,
    }],
})
 
# Historical event from January 2025
client.ingest({
    "run_id": "my-run",
    "items": [{
        "text": "Server migration to AWS completed with zero downtime.",
        "intent": "fact",
        "occurrence_time": 1736899200,  # Jan 15 2025 UTC
    }],
})

Temporal queries

Use min_timestamp and max_timestamp to filter evidence to a specific time window. The filter checks occurrence_time first, falling back to ingestion time.

# "What happened in January 2025?"
results = client.query(
    run_id="my-run",
    query="What technical changes were made?",
    min_timestamp=1735689600,   # Jan 1 2025
    max_timestamp=1738367999,   # Jan 31 2025
)
 
for evidence in results["evidence"]:
    print(f"  {evidence['content'][:80]}")

Without temporal bounds, queries like "What happened last week?" use natural language temporal intent detection and prioritize entries by occurrence time in the recency ranking.

Search budget

The budget parameter controls the depth of retrieval. Use "low" for real-time agents and "high" for accuracy-critical offline analysis.

# Fast retrieval for a real-time chatbot
fast = client.query(run_id="my-run", query="user question", budget="low")
 
# Deep retrieval for a research report
deep = client.query(run_id="my-run", query="comprehensive analysis topic", budget="high")

Staleness detection

When a newer fact contradicts an older one, MuBit marks the older entry as stale and deprioritizes it in ranking. The staleness metadata is available in evidence responses.

results = client.recall(query="Where is the office?")
for evidence in results["evidence"]:
    stale = evidence.get("is_stale", False)
    status = " [STALE]" if stale else ""
    print(f"  {evidence['content'][:60]}{status}")

Mental models

The mental_model entry type stores consolidated entity summaries that are prioritized over raw facts in context assembly. Use this for entities your agent tracks over time.

client.remember(
    content="Alice Chen is a senior engineer specializing in distributed systems. "
            "She prefers async communication and reviews PRs within 24 hours.",
    intent="mental_model",
    metadata={"entity": "alice chen", "consolidated": True},
)

Mental models are returned with higher priority than individual facts in recall() and get_context(). Update them periodically as your agent learns more about an entity.

Failure modes and troubleshooting

Next steps

• See the concrete helper flow at Getting started.
• See the wire contract at Control HTTP reference.