Introduction

UMA is a production-first memory runtime for developers building AI agents. It combines raw evidence, semantic facts, episodic memory, procedural skills, graph links, profiles, and compiled wiki memory into a single SDK with a clean retrieval interface.

Security is a first-class design principle, not a feature layer. The architecture enforces explicit ownership boundaries, tracks provenance through every write, hashes content for integrity, and scans every memory write for prompt-injection patterns. UMA is the first Apache-2.0 memory runtime with OWASP ASI06 (Memory Poisoning) defense built into the canonical write and read paths — source-trust scoring, injection-pattern detection, quarantine, trust-aware retrieval, and an isolation-by-construction vector contract that addresses LLM08 (Vector and Embedding Weaknesses). See the OWASP coverage table below.

UMA manages memory only. It retrieves context, compiles evidence-backed memory, and maintains rebuildable wiki projections. Your agent owns all reasoning, tools, and final responses.

What is UMA

UMA is a production-first memory architecture inspired by how human memory evolves: capture, association, consolidation, forgetting, and recall. It transforms raw interaction history into compiled, evidence-backed knowledge, then lets agents recursively navigate that knowledge under explicit budgets.

Key capabilities

• Bounded, deterministic retrieval — RLM loops use strict step, action, and time budgets.
• Memory as environment — agents peek into memory via safe, snippet-first APIs instead of loading full context into prompts.
• Explicit ownership contracts — write-facing and promotion-facing paths use explicit primitive ownership fields.
• Memory poisoning defense — every user input and assistant reply is scanned against an OWASP-aligned pattern catalog; high-severity hits are quarantined out of retrieval.
• Hardened file ingestion — MIME consistency checks reject executables masquerading as documents; HTML and Markdown are sanitized of active payloads before chunking.
• Provenance as invariant — memory answers and compiled artifacts remain traceable back to raw chunks.
• Compiled wiki memory — canonical records synthesized over evidence, with drift detection and regeneration.
• Configurable backends — use embedded SQLite + LanceDB or FAISS. Use your favorite LLM provider — OpenAI, Claude, or Ollama.
• SDK-first — UMA manages memory only; your agent controls reasoning, tools, and final responses.
• Multi-language support for injection scanning — Bundled English, French, Spanish, German, and Simplified Chinese YAML catalogs are aligned with [OWASP Agent Memory Guard](https://owasp.org/www-project-agent-memory-guard/).

Installation

By the end of this guide you will have UMA running on the default embedded Lite profile — SQLite for authoritative storage, LanceDB for vector retrieval — with your LLM and embedding provider configured. No external database is required.

Prerequisites

Before you begin, ensure the following are available on your machine:

• Python 3.9 or later — check with python --version
• Git — to clone the repository
• An LLM and embedding provider — UMA supports ollama, openai, and anthropic (LLM only) for the LLM; ollama and openai for embeddings. Have your API key or local endpoint ready before editing uma.yaml.

Step 1 — Clone the repository

git clone https://github.com/fad-schme/UMA.git
cd UMA

Step 2 — Create a virtual environment and install

UMA's base install includes SQLite, LanceDB, and support for the Ollama and OpenAI providers. Everything runs in-process — no separate database service to start.

python -m venv .venv
source .venv/bin/activate   # Windows: .venv\Scripts\activate
pip install -e .

Step 3 — Configure your provider

Open config/uma.yaml and set the embedding and llms.uma sections to match your environment. UMA uses its internal LLM for memory extraction, fact summarization, and snippet refinement — not for generating agent replies.

Ollama (local, no API key needed):

embedding:
  provider: "ollama"
  model: "nomic-embed-text"
  dimension: 768
  config:
    host: "http://localhost:11434"

llms:
  uma:
    provider: "ollama"
    model: "qwen2.5:3b"    # or any model you have pulled
    config:
      host: "http://localhost:11434"

OpenAI:

embedding:
  provider: "openai"
  model: "text-embedding-3-small"
  dimension: 1536
  config:
    api_key: "${OPENAI_API_KEY}"

llms:
  uma:
    provider: "openai"
    model: "gpt-4o-mini"
    config:
      api_key: "${OPENAI_API_KEY}"

Anthropic (LLM only — install the extra first):

pip install -e '.[llm]'

embedding:
  provider: "openai"          # Anthropic is LLM-only; use OpenAI or Ollama for embeddings
  model: "text-embedding-3-small"
  dimension: 1536
  config:
    api_key: "${OPENAI_API_KEY}"

llms:
  uma:
    provider: "anthropic"
    model: "claude-haiku-4-5-20251001"
    config:
      api_key: "${ANTHROPIC_API_KEY}"

Step 4 — Verify the install

Run a quick health check to confirm UMA initialises and can reach your provider:

python - <<'PY'
from uma import UMAMemory
memory = UMAMemory.from_yaml("config/uma.yaml").set_context(agent_id="test")
print(memory.health_check())
PY

A healthy response reports "status": "ok" with per-component checks for each SQL store (db:episodic, db:semantic, db:procedural), each vector index, the LLM, and the embedder. If any component shows "status": "error", the detail field names the exact failure — usually an unreachable provider endpoint or a missing API key.

Optional extras

Where UMA stores data

On first use, UMA creates a .uma/ directory in the working directory where you run it:

The SQL files are the authoritative source of truth. The vector files are a rebuildable retrieval accelerator. If the vector index is ever corrupted or needs to be moved, rebuild it with:

await memory.rebuild_vector_indexes(tenant_id="default")

Running the test suite

pip install -r requirements.txt
PYTHONPATH=. python -m pytest -q

Quick Start

Get a minimal agent with UMA memory running on the default embedded Lite profile.

from uma import UMAMemory, InjectionDetectedError

memory = UMAMemory.from_yaml("config/uma.yaml").set_context(agent_id="my-agent")

## Scan all user input for prompt injection 
## before it reaches storage or an LLM.
scan = memory.scan_user_input(user_msg)
if scan["severity"] == "high":
    # do not forward to LLM, do not call process_turn
    return "I can't process that request."

context = await memory.retrieve_context(
    query_text=user_message,
    user_id="user-123",
    tenant_id="default",
    session_id="session-1",
)

messages = [
    {"role": "system", "content": system_prompt},
    {"role": "system", "content": str(context)},
]
agent_reply = await your_agent_llm_generate(messages)

try:
    await memory.process_turn(
        user_id="user-123",
        user_msg=user_message,
        assistant_reply=agent_reply,
        session_id="session-1",
        tenant_id="default",
    )
except InjectionDetectedError as e:
    print(e.severity)       # "high"
    print(e.matched_rules)  # ["prompt_reset", ...]
    print(e.score)          # numeric scan score
    # surface error, alert, block user — your decision

result = await memory.retrieve_memory(
    query_text=user_message,
    user_id="user-123",
    tenant_id="default",
)

await memory.ingest_document(
    file_path="/data/document.txt",
    owner_id="user-123",
    owner_type="user",
    tenant_id="default",
)

Anthropic SDK

UMA wraps around your existing Anthropic loop. Four calls — scan, retrieve, generate, persist — cover the full memory lifecycle. Nothing in your LLM call changes.

import asyncio
import anthropic
from uma import UMAMemory, InjectionDetectedError

memory = UMAMemory.from_yaml("config/uma.yaml").set_context(agent_id="my-agent")
client = anthropic.Anthropic()

# user_id and session_id come from your application at runtime.
# session_id must be stable across all turns in the same conversation thread.

async def handle_turn(user_id: str, session_id: str, user_msg: str) -> str:
    # 1. Pre-LLM gate — never raises, you decide what to do
    scan = memory.scan_user_input(user_msg)
    if scan["severity"] == "high":
        return "I can't process that request."

    # 2. Pull relevant memory into the prompt
    context = await memory.retrieve_context(
        query_text=user_msg,
        user_id=user_id,
        session_id=session_id,
        # tenant_id defaults to "default"
    )
    system_prompt = build_system_prompt(context)   # your function

    # 3. Your LLM call — UMA does not touch this
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        system=system_prompt,
        messages=[{"role": "user", "content": user_msg}],
    )
    reply = response.content[0].text

    # 4. Persist the turn — defense-in-depth scan runs here too
    try:
        await memory.process_turn(
            user_id=user_id,
            user_msg=user_msg,
            assistant_reply=reply,
            session_id=session_id,
        )
    except InjectionDetectedError as e:
        # Payload slipped Layer 1 but caught at Layer 2.
        # Turn not stored — reply already sent.
        log_security_event(e.matched_rules, e.score)

    return reply

OpenAI

Same four-step pattern as the Anthropic SDK — only the client and model change. UMA is provider-agnostic.

import asyncio
from openai import OpenAI
from uma import UMAMemory, InjectionDetectedError

memory = UMAMemory.from_yaml("config/uma.yaml").set_context(agent_id="my-agent")
client = OpenAI()

async def handle_turn(user_msg: str) -> str:
    # 1. Pre-LLM gate
    scan = memory.scan_user_input(user_msg)
    if scan["severity"] == "high":
        return "I can't process that request."

    # 2. Retrieve context
    context = await memory.retrieve_context(
        query_text=user_msg,
        user_id=user_id,
        session_id=session_id,
        # tenant_id defaults to "default"
    )

    # 3. LLM call
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": build_system_prompt(context)},
            {"role": "user",   "content": user_msg},
        ],
    )
    reply = response.choices[0].message.content

    # 4. Persist
    try:
        await memory.process_turn(
            user_id=user_id,
            user_msg=user_msg,
            assistant_reply=reply,
            session_id=session_id,
        )
    except InjectionDetectedError as e:
        log_security_event(e.matched_rules, e.score)

    return reply

LangGraph

UMA slots into a LangGraph agent as a state-aware retrieval step at the start of the node that calls the LLM. Run scan and retrieve before the model node, run persist after the final response. Your graph edges and conditional logic stay unchanged.

import asyncio
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langchain_anthropic import ChatAnthropic
from uma import UMAMemory, InjectionDetectedError

memory = UMAMemory.from_yaml("config/uma.yaml").set_context(agent_id="my-agent")
llm    = ChatAnthropic(model="claude-opus-4-6")

# user_id and session_id flow in through graph state — set at graph.invoke() time.

class AgentState(TypedDict):
    user_id:    str   # required: set by caller at graph.invoke() time
    session_id: str   # required: stable per conversation thread
    user_msg:   str
    uma_ctx:    dict  # populated by retrieve_node, consumed by llm_node
    reply:      str

async def scan_node(state: AgentState) -> AgentState:
    """Layer-1 injection gate — runs before any LLM call."""
    scan = memory.scan_user_input(state["user_msg"])
    if scan["severity"] == "high":
        return {**state, "reply": "I can't process that request."}
    return state

async def retrieve_node(state: AgentState) -> AgentState:
    """Pull UMA context into state so the LLM node can use it."""
    ctx = await memory.retrieve_context(
        query_text=state["user_msg"],
        user_id=state["user_id"],
        session_id=state["session_id"],
        # tenant_id defaults to "default"
    )
    return {**state, "uma_ctx": ctx}

async def llm_node(state: AgentState) -> AgentState:
    """Standard LangGraph LLM node — reads UMA context from state."""
    system = build_system_prompt(state["uma_ctx"])   # your function
    response = await llm.ainvoke([
        ("system", system),
        ("human",  state["user_msg"]),
    ])
    return {**state, "reply": response.content}

async def persist_node(state: AgentState) -> AgentState:
    """Store the completed turn in UMA memory."""
    try:
        await memory.process_turn(
            user_id=state["user_id"],
            user_msg=state["user_msg"],
            assistant_reply=state["reply"],
            session_id=state["session_id"],   # required, non-empty
        )
    except InjectionDetectedError as e:
        log_security_event(e.matched_rules, e.score)
    return state

def already_replied(state: AgentState) -> str:
    return "end" if state.get("reply") else "retrieve"

# Build graph
builder = StateGraph(AgentState)
builder.add_node("scan",     scan_node)
builder.add_node("retrieve", retrieve_node)
builder.add_node("llm",      llm_node)
builder.add_node("persist",  persist_node)
builder.set_entry_point("scan")
builder.add_conditional_edges("scan", already_replied, {"end": END, "retrieve": "retrieve"})
builder.add_edge("retrieve", "llm")
builder.add_edge("llm",      "persist")
builder.add_edge("persist",  END)
graph = builder.compile()

LangChain

Add UMA as a retrieval step before your chain runs and a persistence step after it returns. Use a RunnableLambda to inject UMA context into the chain input, keeping the chain itself unchanged.

import asyncio
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from uma import UMAMemory, InjectionDetectedError

memory = UMAMemory.from_yaml("config/uma.yaml").set_context(agent_id="my-agent")

llm    = ChatAnthropic(model="claude-opus-4-6")
prompt = ChatPromptTemplate.from_messages([
    ("system", "{system_context}"),
    ("human",  "{user_msg}"),
])
chain  = prompt | llm

async def run_with_memory(user_id: str, session_id: str, user_msg: str) -> str:
    # 1. Scan
    scan = memory.scan_user_input(user_msg)
    if scan["severity"] == "high":
        return "I can't process that request."

    # 2. Retrieve and inject into chain input
    ctx = await memory.retrieve_context(
        query_text=user_msg,
        user_id=user_id,
        session_id=session_id,
        # tenant_id defaults to "default"
    )

    # 3. Run chain
    response = await chain.ainvoke({
        "system_context": build_system_prompt(ctx),
        "user_msg":       user_msg,
    })
    reply = response.content

    # 4. Persist
    try:
        await memory.process_turn(
            user_id=user_id,
            user_msg=user_msg,
            assistant_reply=reply,
            session_id=session_id,
        )
    except InjectionDetectedError as e:
        log_security_event(e.matched_rules, e.score)

    return reply

CrewAI

In CrewAI, UMA memory wraps the task execution boundary. Retrieve before the crew runs, persist after it returns. Each agent in a multi-agent crew uses the same UMAMemory instance but a distinct user_id or session_id to keep episodic lanes separate.

import asyncio
from crewai import Agent, Task, Crew
from uma import UMAMemory, InjectionDetectedError

memory = UMAMemory.from_yaml("config/uma.yaml").set_context(agent_id="crew-agent")

researcher = Agent(
    role="Research Analyst",
    goal="Find and summarise relevant information",
    backstory="You are a thorough researcher.",
    llm="claude-opus-4-6",
)

async def run_crew_with_memory(user_id: str, session_id: str, topic: str) -> str:
    # 1. Scan
    scan = memory.scan_user_input(topic)
    if scan["severity"] == "high":
        return "I can't process that request."

    # 2. Retrieve — inject prior knowledge into the task description
    ctx = await memory.retrieve_context(
        query_text=topic,
        user_id=user_id,
        session_id=session_id,
        # tenant_id defaults to "default"
    )
    prior = format_context_for_task(ctx)   # your function

    # 3. Build and run crew
    task = Task(
        description=f"Research: {topic}\n\nPrior context:\n{prior}",
        expected_output="A concise research summary.",
        agent=researcher,
    )
    crew   = Crew(agents=[researcher], tasks=[task])
    result = crew.kickoff()
    reply  = str(result)

    # 4. Persist
    try:
        await memory.process_turn(
            user_id=user_id,
            user_msg=topic,
            assistant_reply=reply,
            session_id=session_id,
        )
    except InjectionDetectedError as e:
        log_security_event(e.matched_rules, e.score)

    return reply

PydanticAI

Use PydanticAI's system_prompt dependency to inject UMA context, and a post-run hook to persist the turn. The agent definition stays clean — UMA is wired at the run boundary, not inside the agent logic.

import asyncio
from dataclasses import dataclass
from pydantic_ai import Agent, RunContext
from uma import UMAMemory, InjectionDetectedError

memory = UMAMemory.from_yaml("config/uma.yaml").set_context(agent_id="my-agent")

@dataclass
class UMADeps:
    uma_context: dict   # populated before agent.run()

agent = Agent(
    "claude-opus-4-6",
    deps_type=UMADeps,
    system_prompt="You are a helpful assistant.",
)

@agent.system_prompt
async def inject_memory(ctx: RunContext[UMADeps]) -> str:
    """Injects UMA context into the system prompt at run time."""
    return build_system_prompt(ctx.deps.uma_context)   # your function

async def handle_turn(user_msg: str) -> str:
    # 1. Scan
    scan = memory.scan_user_input(user_msg)
    if scan["severity"] == "high":
        return "I can't process that request."

    # 2. Retrieve
    ctx = await memory.retrieve_context(
        query_text=user_msg,
        user_id=user_id,
        session_id=session_id,
        # tenant_id defaults to "default"
    )

    # 3. Run agent with UMA context as dependency
    result = await agent.run(user_msg, deps=UMADeps(uma_context=ctx))
    reply  = result.data

    # 4. Persist
    try:
        await memory.process_turn(
            user_id=user_id,
            user_msg=user_msg,
            assistant_reply=reply,
            session_id=session_id,
        )
    except InjectionDetectedError as e:
        log_security_event(e.matched_rules, e.score)

    return reply

Custom loop

Not using a framework? UMA is a Python library. Call the four methods directly in whatever execution loop you are running — HTTP handler, CLI, async queue worker, or a raw asyncio loop.

import asyncio
from uma import UMAMemory, InjectionDetectedError

memory = UMAMemory.from_yaml("config/uma.yaml").set_context(agent_id="my-agent")

async def agent_loop(user_id: str, session_id: str, user_msg: str) -> str:

    # ── Step 1: pre-LLM gate ────────────────────────────────────────────
    # Advisory, synchronous, never raises. You decide what to do with the result.
    scan = memory.scan_user_input(user_msg)
    if scan["severity"] == "high":
        return "I can't process that request."

    # ── Step 2: retrieve context ─────────────────────────────────────────
    # Pulls ranked evidence from all active memory lanes and merges it.
    # Use lane_filter to narrow: ["raw","semantic"] for RAG-only,
    # ["working_memory","episodic"] for continuity-only.
    context = await memory.retrieve_context(
        query_text=user_msg,
        user_id=user_id,
        session_id=session_id,
        # tenant_id defaults to "default"
    )

    # ── Step 3: your LLM call ────────────────────────────────────────────
    # UMA does not call the LLM for you. Build your prompt however you like.
    system_prompt = build_system_prompt(context)   # your function
    reply = await your_llm(system_prompt, user_msg)

    # ── Step 4: persist the turn ─────────────────────────────────────────
    # Stores episode, extracts semantic facts, updates working memory.
    # Defense-in-depth: rescans user_msg before writing anything.
    # Raises InjectionDetectedError on high severity — nothing is stored.
    try:
        await memory.process_turn(
            user_id=user_id,
            user_msg=user_msg,
            assistant_reply=reply,
            session_id=session_id,
        )
    except InjectionDetectedError as e:
        # Layer 2 caught what Layer 1 missed — or skip_scan was used.
        # The turn was NOT stored. Surface the event; the reply already went out.
        log_security_event({
            "rules":    e.matched_rules,   # e.g. ["prompt_reset", "role_impersonation"]
            "score":    e.score,
            "severity": e.severity,        # always "high" when this raises
        })

    return reply


# ── Ingest a document into the knowledge base ─────────────────────────────
# Do this once per document. After ingest, retrieve_context surfaces its chunks.
async def ingest_doc(file_path: str):
    report = await memory.ingest_document(
        file_path=file_path,
        owner_type="user",          # or "agent" / "workspace"
        owner_id=user_id,
    )
    print(report)   # IngestReport with chunk count, warnings, manifest status

Framework cheat sheet

MCP Server

UMA ships an MCP server at mcp/server.py that exposes its five core operations as MCP tools over stdio. Connect it to Claude Desktop, Claude Code, Cursor, or any MCP-compatible client and your AI assistant can retrieve context, recall compiled memory, store conversation turns, and ingest documents — all backed by UMA's ownership-scoped, trust-gated memory architecture.

Prerequisites

Before connecting any client, ensure you have:

• Python 3.9+ with UMA installed: pip install -e . from the repo root
• The MCP library: pip install "mcp[cli]>=1.0.0"
• A valid config/uma.yaml with your LLM and embedding provider configured

pip install -e .
pip install "mcp[cli]>=1.0.0"

Environment variables

The server is configured entirely through environment variables — no arguments, no flags. These are set in the client config and passed to the server process at launch.

Client setup

Pick your client below. In every case, use the absolute path to your repo clone — relative paths are not reliable when the client launches the server process. After saving the config, fully quit and relaunch the client for changes to take effect.

Claude Desktop

Edit the Claude Desktop config file for your platform:

{
  "mcpServers": {
    "uma": {
      "command": "python",
      "args": ["/absolute/path/to/uma/mcp/server.py"],
      "env": {
        "UMA_CONFIG_PATH": "/absolute/path/to/uma/config/uma.yaml",
        "UMA_AGENT_ID": "agent-default",
        "PYTHONPATH": "/absolute/path/to/uma"
      }
    }
  }
}

Claude Code

Register the server from your terminal with claude mcp add. Claude Code uses stdio transport, so pass the server script directly:

claude mcp add uma \
  --command python \
  --args "/absolute/path/to/uma/mcp/server.py" \
  --env UMA_CONFIG_PATH=/absolute/path/to/uma/config/uma.yaml \
  --env UMA_AGENT_ID=agent-default \
  --env PYTHONPATH=/absolute/path/to/uma

Verify the server is registered and its tools are visible:

claude mcp list
claude mcp get uma

Cursor

Add to your global MCP config at ~/.cursor/mcp.json, or per-project at .cursor/mcp.json in the project root:

{
  "mcpServers": {
    "uma": {
      "command": "python",
      "args": ["/absolute/path/to/uma/mcp/server.py"],
      "env": {
        "UMA_CONFIG_PATH": "/absolute/path/to/uma/config/uma.yaml",
        "UMA_AGENT_ID": "agent-default",
        "PYTHONPATH": "/absolute/path/to/uma"
      }
    }
  }
}

Alternatively, go to Cursor Settings → MCP and add a new server entry with the command, args, and environment variables above.

Verify the connection

Before connecting a client, run the server directly from your terminal to confirm it starts without errors:

UMA_CONFIG_PATH=/absolute/path/to/uma/config/uma.yaml \
UMA_AGENT_ID=agent-default \
  python mcp/server.py

The server speaks JSON-RPC over stdio. On a healthy start you will see no output — it waits for a client to connect. If your configured LLM or embedding provider is unreachable, the error appears in stderr with the connection detail; confirm your provider is running and that uma.yaml points to the correct host or API endpoint.

Once connected through a client, ask your assistant:

Call the uma health_check tool and show me the result.

A healthy response looks like {"status": "ok", "stores": {...}, "embedder": "ok"}. If any component reports an error, check that your configured provider is reachable and that UMA_CONFIG_PATH points to a valid uma.yaml.

Available Tools

The MCP server exposes five tools. Each maps directly to a UMAMemory method — no translation layer, no hidden logic. Every tool returns a JSON string.

Tool parameters

retrieve_context

retrieve_memory

process_turn

ingest_document

Recommended flow

For a standard conversation with persistent memory, call the tools in this order on every turn:

MCP Troubleshooting

Security by Design

UMA is the first Apache-2.0 memory runtime where OWASP ASI06 (memory poisoning) defense lives in the canonical write and read paths — not bolted on as middleware. Every memory write carries provenance, a source-trust score, and a content integrity hash. Suspicious content is scanned at ingest, quarantined out of retrieval, and reviewable through a small management API. Retrieval itself is trust-aware.

The defenses are part of the architecture, not optional features layered on top. They run on every turn, every document ingest, every bootstrap import, with no caller code required to activate them.

The seven primitives

Two-layer input scanning

For conversation input, UMA exposes a pre-LLM gate and a defense-in-depth write-time scan. Call scan_user_input at the top of your agent loop before any LLM call. It returns a result dict and never raises — the caller decides what to do. process_turn rescans the same input at storage time and raises InjectionDetectedError on high severity, so nothing poisoned reaches working memory, episodic store, or fact extraction.

from uma import UMAMemory, InjectionDetectedError

memory = UMAMemory.from_yaml("config/uma.yaml")

# Layer 1 — pre-LLM gate
scan = memory.scan_user_input(user_msg)
if scan["severity"] == "high":
    return "I can't process that request."

# ... memory.retrieve_context(...) ...
# ... your LLM call ...

# Layer 2 — defense in depth at write time
try:
    await memory.process_turn(
        user_id="user-123",
        user_msg=user_msg,
        assistant_reply=reply,
        session_id="session-1",
    )
except InjectionDetectedError as e:
    print(e.severity, e.matched_rules, e.score)
          

Severity to behavior

Managing quarantined records

High-severity scans at the document or chunk level set quarantined_at and preserve the record outside retrieval. Operators review through the management API.

from uma.api.management import (
    list_quarantined,
    reinstate_quarantined,
    purge_quarantined,
    verify_integrity,
)

# Review what was quarantined
records = await list_quarantined(memory, owner_type="agent", owner_id="agent-1")

# Restore a false positive
await reinstate_quarantined(
    memory,
    record_id="...",
    lane="semantic",
    owner_type="agent",
    owner_id="agent-1",
    reason="reviewed: legitimate quote from research paper",
)

# Permanently remove a confirmed attack
await purge_quarantined(memory, record_id="...", lane="semantic",
                       owner_type="agent", owner_id="agent-1",
                       reason="confirmed injection payload")

# On-demand integrity check; mismatch quarantines the record
result = await verify_integrity(memory, record_id="...", lane="semantic",
                                owner_type="agent", owner_id="agent-1")
          

OWASP Coverage

UMA is a memory SDK — not every OWASP category applies. The mapping below is the honest accounting of what UMA covers and what belongs to the calling application. UMA addresses six of the ten OWASP Top 10 for LLM Applications 2025 categories and three of the ten Agentic AI Security (ASI) categories.

Retrieval Products

UMA exposes two distinct retrieval products on UMAMemory.

Both paths run through a small lane-aware planner that decides which canonical lanes participate, surfaces excluded lanes and reasons in trace data, and keeps backend mechanics below that boundary.

# RAG-style context for the LLM
context = await memory.retrieve_context(
   query_text=user_message,
   user_id="user-123"
)

# Compiled memory with provenance for continuity
result = await memory.retrieve_memory(
    query_text=user_message,
    user_id="user-123"
)

Ingest Stages

Document ingest is split into three explicit internal stages. Each stage is independent and rerunnable without repeating prior stages — and each stage has its own security boundary, so untrusted input is validated, scanned, and trust-scored before it ever reaches a retrieval query.

Security primitives at the ingest boundary

Five defenses run before any ingested content can be retrieved. Each maps to an OWASP Agentic Security Initiative control.

Memory Lanes

UMA organizes all memory into canonical lanes. Each lane has its own storage, retrieval semantics, and ownership rules.

Ownership Boundaries

UMA enforces a first-class logical separation between an agent's global knowledge and user-specific memory. This separation is enforced through explicit ownership metadata across SQL storage, vector embeddings, graph nodes and edges, retrieval filters, write paths, and promotion paths.

Retrieval searches the appropriate scope first, merges results deterministically across scopes, and applies promotion or demotion policies when knowledge should move between layers.

Compiled Wiki Memory

UMA treats wiki pages as managed memory records, not markdown files. Canonical wiki state lives in UMA records with kind="wiki_page" and kb_lane="wiki".

What uma.memory.wiki manages

• Page identity and slugging
• Lifecycle status and evidence links
• Deterministic updates and drift checks
• Markdown projection and page regeneration

Markdown under wiki/*.md is projection-only output. Wiki pages are synthesized views over evidence — they are not terminal truth and can be deleted and rebuilt from the underlying evidence at any time.

Management API

from uma.api.management import (
    update_wiki_page,
    export_wiki_projection,
    lint_memory_drift,
)

# Lint stale, unsupported, or conflicted wiki state
report = await lint_memory_drift(memory)

# Export a rebuildable markdown projection
await export_wiki_projection(
    memory,
    compiled_answer,
    output_path="wiki/example.md",
)

Graph Memory

The graph is the lane that connects stored memories to each other. Facts, episodes, and entities are nodes; the relationships between them are edges. When a retrieval pulls a single fact, the graph lets UMA follow its connections — "what else is linked to this person, this project, this decision?" — and bring back related memories the user didn't ask for directly.

This is how UMA recalls context that lives between documents and turns. A user mentioning a project name can surface the people associated with it, the decisions made about it, and the prior conversations that touched it — even when none of that text matches the query lexically or semantically.

What lives in the graph

Edges are predicate-scoped, meaning every connection carries a typed relationship label (works_on, mentioned_in, decided_by, etc.) rather than a generic "is related to". Predicates make traversal precise: you can expand along the relationships that matter for the current query and ignore the rest.

• Facts and episodes are connected over time. When a new fact references an entity that already exists in the graph, an edge is added — so the graph grows denser as the agent learns more, without rewriting what was already there.
• Edges carry temporal and provenance metadata. When the relationship was first established, when it was last confirmed, which fact and which source chunk it came from, and who owns it.
• Ownership flows through every edge. A fact → graph edge inherits owner_type and owner_id from its source fact. Graph traversal cannot cross tenant boundaries.

Graph neighbor queries

When retrieval activates the graph, it walks outward from a seed node — typically a fact already surfaced by vector or lexical search — and pulls in neighboring nodes along chosen predicates. Depth and result limits cap every traversal so a single query can't fan out across the whole graph.

Configuration

config/uma.yaml is the committed default. It runs the embedded Lite profile — SQLite for storage, LanceDB for vectors, no external services. Edit it to point at your LLM and embedding providers, then load it at runtime:

memory = UMAMemory.from_yaml("config/uma.yaml")

UMA ships with support for OpenAI, Anthropic, and Ollama as LLM providers, and OpenAI or Ollama for embeddings. Set the provider and model in the config:

llms:
  uma:
    provider: "ollama"
    model: "llama3"

embedding:
  provider: "ollama"
  model: "nomic-embed-text"
  dimension: 1536

Logging

UMA logs to stdout and to a file by default. Override with environment variables:

Maintenance

The management API at uma.api.management covers the day-to-day operational surface — quarantine review, integrity checks, retrieval auditing, and result explanation. Everything here is opt-in: call it when you need it, ignore it otherwise. None of it sits in the hot path of retrieve_context, retrieve_memory, or process_turn.

Quarantine review

High-severity injection scans don't delete — they quarantine. Artifacts are stored with a quarantined_at timestamp and excluded from retrieval, but the rows stay in the database so you can review and decide.

from uma.api.management import list_quarantined, reinstate_quarantined, purge_quarantined

# Review what's been quarantined
rows = await list_quarantined(memory, lane="semantic")

# Restore a false positive
await reinstate_quarantined(memory, record_id="fact-abc", lane="semantic")

# Permanently delete confirmed bad content
await purge_quarantined(memory, record_id="chunk-xyz", lane="raw")

Integrity verification

verify_integrity recomputes the canonical content hash for any stored Fact, Episode, Skill, or Chunk and compares it to the hash recorded at write time. A match returns status="verified" without mutating anything; a mismatch quarantines the record and returns status="failed" with the diff.

from uma.api.management import verify_integrity, lint_memory_drift

# Verify a single record on demand
result = await verify_integrity(memory, record_id="fact-abc", lane="semantic")

# Batch-check typed-lane artifacts
drift = await lint_memory_drift(memory, artifact, user_id="user-123")

lint_memory_drift routes typed-lane artifacts through verify_integrity automatically, so you can run batch checks without calling the function directly. Background scanning across the full dataset is an Enterprise capability and is not part of this SDK — in UMA Lite you trigger checks on demand.

Result explanation

Retrieval is supposed to be inspectable. explain_result returns the lane plan, candidate pool size, fusion ordering, and trust-adjusted scores for a given retrieval result — useful for debugging unexpected rankings or showing reviewers exactly why an answer surfaced what it did.

from uma.api.management import explain_result

result = await memory.retrieve_memory(query_text=q, user_id="user-123", ...)
explanation = await explain_result(memory, result, user_id="user-123")

Retrieval audit log

Every retrieval call is also recorded in an audit log — hashed query preview, scope, severity, result counts — queryable via list_retrieval_audit. Disable it by setting security.retrieval_audit_enabled: false in your YAML.

Health and index rebuild

Two more operational primitives live directly on UMAMemory: health_check() returns the runtime status of each lane and backend, and rebuild_vector_indexes() regenerates the LanceDB indexes from SQL (the authoritative source) — useful after a schema migration or if a vector store becomes corrupt.

# Status snapshot
status = memory.health_check()

# Rebuild vector indexes from SQL
await memory.rebuild_vector_indexes(tenant_id="default")

Interactive Documentation — Ask, Don't Read

You don't need to read tons of documentation to use UMA. Ask your coding assistant instead.

UMA ships interactive documentation as a set of eight reference files under .claude/skills/. Each file follows the Agent Skills specification, which means Claude Code (and any Agent Skills-compatible assistant) automatically loads the relevant file when its description matches your question. No setup, no @ mentions, no manual context-pasting. Just ask.

Ask, don't read

# "How do I integrate UMA into my chatbot?"
→ uma-agent-loop.md loads — end-to-end pattern with code

# "What happens when a user sends a prompt injection?"
→ uma-security.md + uma-quarantine.md load — full flow from scan to storage

# "How do I write a custom vector backend?"
→ uma-vector-contract.md loads — the contract, atomicity, score normalization

# "How do I filter by lane?"
→ uma-lanes.md loads — the six lanes, when to use each

# "Can you help me configure Anthropic as the LLM?"
→ uma-configure.md loads — full YAML reference

The eight reference files

Each file is under 500 lines, follows the Agent Skills specification, and is verified against the codebase — no phantom APIs. Assistants that don't follow .claude/skills/ can read the same files directly, or via a symlink at .agents/skills/ if your tooling uses that path.

UMA contribution

UMA is a library, not a hosted service, and cannot itself hold compliance certifications. UMA's security primitives are designed to contribute to compliance programs in customers building on top of it.
The architecture supports work toward OWASP ASI baseline, NIST AI RMF, ISO/IEC 42001 and 27001, SOC 2 Trust Services Criteria, GDPR (notably Articles 17, 30, 32), CCPA, and the EU AI Act, by providing the technical controls and evidence artifacts (ownership scoping, audit events, integrity verification, deletion attestation) that compliance programs require.
Customers remain responsible for their overall control environment, audit posture, and certification scope.