Python SDK

from membrane import MembraneClient, Sensitivity, TrustContext

# Connect to the running Membrane daemon
client = MembraneClient("localhost:9090")

# Ingest an event
record = client.ingest_event(
    event_kind="file_edit",
    ref="src/main.py",
    summary="Refactored authentication module",
    sensitivity=Sensitivity.LOW,
)
print(f"Created record: {record.id}")

# Retrieve memories relevant to a task
trust = TrustContext(
    max_sensitivity=Sensitivity.MEDIUM,
    authenticated=True,
    actor_id="agent-1",
)
records = client.retrieve("fix the login bug", trust=trust, limit=5)
for r in records:
    print(f"  [{r.type.value}] {r.id} (salience={r.salience:.2f})")

# Reinforce a useful memory
client.reinforce(record.id, actor="agent-1", rationale="Used successfully")

# Clean up
client.close()

with MembraneClient("localhost:9090") as client:
    record = client.ingest_observation(
        subject="user",
        predicate="prefers",
        obj={"language": "Python"},
        sensitivity=Sensitivity.LOW,
    )

MembraneClient(
    addr: str = "localhost:9090",
    *,
    tls: bool = False,
    tls_ca_cert: str | None = None,
    api_key: str | None = None,
    timeout: float | None = None,
)

client = MembraneClient(
    "membrane.example.com:443",
    tls=True,                        # use TLS transport
    tls_ca_cert="/path/to/ca.pem",   # optional custom CA
    api_key="your-api-key",          # Bearer token auth
    timeout=10.0,                    # default timeout in seconds
)

def ingest_event(
    self,
    event_kind: str,
    ref: str,
    *,
    summary: str = "",
    sensitivity: Sensitivity | str = Sensitivity.LOW,
    source: str = "python-client",
    tags: Sequence[str] | None = None,
    scope: str = "",
    timestamp: str | None = None,
) -> MemoryRecord

record = client.ingest_event(
    source="my-agent",
    event_kind="tool_call",
    ref="task#1",
    summary="Ran database migration successfully",
    tags=["db", "migration"],
)

def ingest_tool_output(
    self,
    tool_name: str,
    *,
    args: dict[str, Any] | None = None,
    result: Any = None,
    sensitivity: Sensitivity | str = Sensitivity.LOW,
    source: str = "python-client",
    depends_on: Sequence[str] | None = None,
    tags: Sequence[str] | None = None,
    scope: str = "",
    timestamp: str | None = None,
) -> MemoryRecord

record = client.ingest_tool_output(
    "run_tests",
    args={"suite": "auth"},
    result={"passed": 42, "failed": 0},
    tags=["tests", "auth"],
)

def ingest_observation(
    self,
    subject: str,
    predicate: str,
    obj: Any,
    *,
    sensitivity: Sensitivity | str = Sensitivity.LOW,
    source: str = "python-client",
    tags: Sequence[str] | None = None,
    scope: str = "",
    timestamp: str | None = None,
) -> MemoryRecord

record = client.ingest_observation(
    subject="user",
    predicate="prefers",
    obj={"language": "Python"},
    sensitivity=Sensitivity.LOW,
)

def ingest_outcome(
    self,
    target_record_id: str,
    outcome_status: OutcomeStatus | str,
    *,
    source: str = "python-client",
    timestamp: str | None = None,
) -> MemoryRecord

from membrane import OutcomeStatus

client.ingest_outcome(record.id, OutcomeStatus.SUCCESS)

def ingest_working_state(
    self,
    thread_id: str,
    state: str,
    *,
    next_actions: Sequence[str] | None = None,
    open_questions: Sequence[str] | None = None,
    context_summary: str = "",
    active_constraints: Sequence[dict[str, Any]] | None = None,
    sensitivity: Sensitivity | str = Sensitivity.LOW,
    source: str = "python-client",
    tags: Sequence[str] | None = None,
    scope: str = "",
    timestamp: str | None = None,
) -> MemoryRecord

client.ingest_working_state(
    thread_id="session-001",
    state="executing",
    next_actions=["run tests", "deploy"],
    context_summary="Backend initialized, frontend pending, docs TODO",
)

def retrieve(
    self,
    task_descriptor: str,
    *,
    trust: TrustContext | None = None,
    memory_types: Sequence[MemoryType | str] | None = None,
    min_salience: float = 0.0,
    limit: int = 10,
) -> list[MemoryRecord]

trust = TrustContext(
    max_sensitivity=Sensitivity.MEDIUM,
    authenticated=True,
    actor_id="agent-1",
)
records = client.retrieve(
    "database operations",
    trust=trust,
    memory_types=["semantic", "competence"],
    limit=5,
)

def retrieve_with_selection(
    self,
    task_descriptor: str,
    *,
    trust: TrustContext | None = None,
    memory_types: Sequence[MemoryType | str] | None = None,
    min_salience: float = 0.0,
    limit: int = 10,
) -> RetrieveResult

@dataclass
class RetrieveResult:
    records: list[MemoryRecord]
    selection: SelectionResult | None  # optional selector metadata

@dataclass
class SelectionResult:
    selected: list[MemoryRecord]
    confidence: float
    needs_more: bool

def retrieve_by_id(
    self,
    record_id: str,
    *,
    trust: TrustContext | None = None,
) -> MemoryRecord

record = client.retrieve_by_id("rec-uuid-here")

def supersede(
    self,
    old_id: str,
    new_record: dict[str, Any] | MemoryRecord,
    actor: str,
    rationale: str,
) -> MemoryRecord

updated = client.supersede(
    old_record.id,
    {**old_record.to_dict(), "payload": {"version": "2.0"}},
    actor="agent-1",
    rationale="Updated version field",
)

def fork(
    self,
    source_id: str,
    forked_record: dict[str, Any] | MemoryRecord,
    actor: str,
    rationale: str,
) -> MemoryRecord

variant = client.fork(
    source_record.id,
    {**source_record.to_dict(), "scope": "dev"},
    actor="agent-1",
    rationale="Different behavior for dev environment",
)

def retract(
    self,
    record_id: str,
    actor: str,
    rationale: str,
) -> None

client.retract(record.id, actor="agent-1", rationale="No longer accurate")

def merge(
    self,
    record_ids: Sequence[str],
    merged_record: dict[str, Any] | MemoryRecord,
    actor: str,
    rationale: str,
) -> MemoryRecord

merged = client.merge(
    [id1, id2, id3],
    {"type": "semantic", "payload": {"fact": "combined knowledge"}},
    actor="agent-1",
    rationale="Consolidating duplicate semantic records",
)

def contest(
    self,
    record_id: str,
    contesting_ref: str,
    actor: str,
    rationale: str,
) -> None

client.contest(
    record.id,
    conflicting_record.id,
    actor="agent-1",
    rationale="New evidence contradicts this",
)

def reinforce(
    self,
    record_id: str,
    actor: str,
    rationale: str,
) -> None

client.reinforce(record.id, actor="agent-1", rationale="Used successfully in task")

def penalize(
    self,
    record_id: str,
    amount: float,
    actor: str,
    rationale: str,
) -> None

client.penalize(record.id, amount=0.2, actor="agent-1", rationale="Led to incorrect result")

def get_metrics(self) -> dict[str, Any]

metrics = client.get_metrics()
print(metrics["total_records"], metrics["avg_salience"])

def close(self) -> None

class Sensitivity(str, Enum):
    PUBLIC = "public"
    LOW    = "low"
    MEDIUM = "medium"
    HIGH   = "high"
    HYPER  = "hyper"

@dataclass
class TrustContext:
    max_sensitivity: Sensitivity = Sensitivity.LOW  # highest sensitivity the caller may access
    authenticated: bool = False
    actor_id: str = ""
    scopes: list[str] = field(default_factory=list)

@dataclass
class MemoryRecord:
    id: str
    type: MemoryType           # "episodic" | "working" | "semantic" | "competence" | "plan_graph"
    sensitivity: Sensitivity
    confidence: float          # 0–1 applicability score
    salience: float            # 0–1 current importance score
    scope: str
    tags: list[str]
    created_at: str
    updated_at: str
    lifecycle: Lifecycle | None
    provenance: Provenance | None
    relations: list[Relation]
    payload: Any               # type-specific structured payload
    audit_log: list[AuditEntry]

class MemoryType(str, Enum):
    EPISODIC   = "episodic"
    WORKING    = "working"
    SEMANTIC   = "semantic"
    COMPETENCE = "competence"
    PLAN_GRAPH = "plan_graph"

class OutcomeStatus(str, Enum):
    SUCCESS = "success"
    FAILURE = "failure"
    PARTIAL = "partial"

@dataclass
class Relation:
    target_id: str = ""   # ID of the related MemoryRecord
    kind: str = ""        # relationship type (maps to Go Predicate)
    weight: float = 1.0   # strength of the relationship [0, 1]

from membrane import (
    AuditAction,     # Enum: create, revise, fork, merge, delete, reinforce, decay
    AuditEntry,      # Dataclass: action, actor, timestamp, rationale
    DecayCurve,      # Enum: EXPONENTIAL = "exponential"
    DecayProfile,    # Dataclass: curve, half_life_seconds
    DeletionPolicy,  # Enum: AUTO_PRUNE, MANUAL_ONLY, NEVER
    EdgeKind,        # Enum: DATA = "data", CONTROL = "control"
    Lifecycle,       # Dataclass: decay, last_reinforced_at, deletion_policy
    ProvenanceKind,  # Enum: EVENT, ARTIFACT, TOOL_CALL, OBSERVATION, OUTCOME
    ProvenanceSource,# Dataclass: kind, ref, timestamp
    Provenance,      # Dataclass: sources (list[ProvenanceSource])
    Relation,        # Dataclass: target_id, kind, weight
    RetrieveResult,  # Dataclass: records, selection
    RevisionStatus,  # Enum: ACTIVE, CONTESTED, RETRACTED
    SelectionResult, # Dataclass: selected, confidence, needs_more
    TaskState,       # Enum: PLANNING, EXECUTING, BLOCKED, WAITING, DONE
    ValidityMode,    # Enum: GLOBAL, CONDITIONAL, TIMEBOXED
)

import grpc
from membrane import MembraneClient

client = MembraneClient("localhost:9090")
try:
    record = client.retrieve_by_id("nonexistent-id")
except grpc.RpcError as e:
    print(e.code())     # e.g. StatusCode.NOT_FOUND
    print(e.details())  # human-readable message
finally:
    client.close()