Python SDK

The Python SDK communicates with membraned over gRPC using the typed membrane.v1.MembraneService protobuf contract. Arbitrary capture content uses google.protobuf.Value; memory records, graph nodes, graph edges, and response envelopes are typed protobuf messages converted into Python dataclasses.

Installation

pip install -e clients/python

For development:

pip install -e clients/python[dev]

Quick Start

from membrane import MembraneClient, Sensitivity, SourceKind, TrustContext

client = MembraneClient("localhost:9090")

capture = client.capture_memory(
    {
        "ref": "src/auth.py",
        "text": "Refactored auth middleware and verified package tests",
        "file": "src/auth.py",
    },
    source_kind=SourceKind.EVENT,
    reason_to_remember="Keep the auth refactor available for future debugging",
    summary="Refactored auth middleware",
    sensitivity=Sensitivity.LOW,
    tags=["auth", "python"],
)
print(capture.primary_record.id)

trust = TrustContext(
    max_sensitivity=Sensitivity.MEDIUM,
    authenticated=True,
    actor_id="agent-1",
)

graph = client.retrieve_graph(
    "debug auth",
    trust=trust,
    memory_types=["entity", "semantic", "competence", "episodic"],
    root_limit=5,
    node_limit=20,
    max_hops=1,
)

for node in graph.nodes:
    print(f"[{node.record.type.value}] {node.record.id} root={node.root} hop={node.hop}")

client.close()

Client Options

client = MembraneClient(
    "membrane.example.com:443",
    tls=True,
    tls_ca_cert="/path/to/ca.pem",
    api_key="your-api-key",
    timeout=10.0,
)

addrstrdefault: localhost:9090

gRPC address of the Membrane daemon.

tlsbool

Use TLS transport.

tls_ca_certstr

Optional CA certificate path. Supplying it implies TLS.

api_keystr

Bearer token sent in gRPC metadata.

timeoutfloat

Default timeout in seconds for RPC calls.

The client also supports the context-manager protocol:

with MembraneClient("localhost:9090") as client:
    capture = client.capture_memory({"text": "Remember this"})

---

Capture

def capture_memory(
    content: Any,
    *,
    source_kind: SourceKind | str = SourceKind.AGENT_TURN,
    context: Any = None,
    reason_to_remember: str = "",
    proposed_type: MemoryType | str | None = None,
    summary: str = "",
    sensitivity: Sensitivity | str = Sensitivity.LOW,
    source: str = "python-client",
    tags: Sequence[str] | None = None,
    scope: str = "",
    timestamp: str | None = None,
) -> CaptureMemoryResult

Captures a rich memory candidate. content must be JSON-compatible.

capture = client.capture_memory(
    {
        "tool_name": "go test",
        "args": {"packages": ["./pkg/auth"]},
        "result": {"exit_code": 0},
    },
    source_kind=SourceKind.TOOL_OUTPUT,
    source="auth-agent",
    reason_to_remember="Successful auth package verification",
    summary="Auth package tests passed",
    tags=["auth", "tests"],
    scope="project-auth",
    sensitivity=Sensitivity.LOW,
)

print(capture.primary_record.id)
print(len(capture.created_records))
print(len(capture.edges))

CaptureMemoryResult contains:

Field	Type
primary_record	MemoryRecord
created_records	list[MemoryRecord]
edges	list[GraphEdge]

---

Retrieve Graph

def retrieve_graph(
    task_descriptor: str,
    *,
    trust: TrustContext | None = None,
    memory_types: Sequence[MemoryType | str] | None = None,
    min_salience: float = 0.0,
    root_limit: int = 10,
    node_limit: int = 25,
    edge_limit: int = 100,
    max_hops: int = 1,
) -> RetrieveGraphResult

Retrieves ranked root memories and a bounded graph neighborhood.

graph = client.retrieve_graph(
    "debug auth retry failures",
    trust=TrustContext(
        max_sensitivity=Sensitivity.MEDIUM,
        authenticated=True,
        actor_id="auth-agent",
        scopes=["project-auth"],
    ),
    memory_types=["entity", "semantic", "competence", "episodic"],
    root_limit=8,
    node_limit=20,
    edge_limit=80,
    max_hops=1,
)

root_records = [node.record for node in graph.nodes if node.root]

RetrieveGraphResult contains:

Field	Type
nodes	list[GraphNode]
edges	list[GraphEdge]
root_ids	list[str]
selection	`SelectionResult

---

Retrieve By ID

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

record = client.retrieve_by_id(
    record_id,
    trust=TrustContext(
        max_sensitivity=Sensitivity.HIGH,
        authenticated=True,
        actor_id="support-agent",
        scopes=["project-auth"],
    ),
)

---

Revision

client.supersede(old_id, new_record, "agent", "updated fact")
client.fork(source_id, forked_record, "agent", "different in staging")
client.retract(record_id, "agent", "no longer accurate")
client.merge([id1, id2], merged_record, "agent", "deduplicated")
client.contest(record_id, conflicting_ref, "agent", "conflicting evidence")

new_record, forked_record, and merged_record may be dictionaries matching the MemoryRecord wire shape or MemoryRecord dataclasses.

Reinforcement

client.reinforce(record_id, "agent", "used successfully")
client.penalize(record_id, 0.2, "agent", "not useful for this task")

Metrics

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

---

Types And Enums

from membrane import (
    BUILTIN_ENTITY_TYPES,
    EntityPayload,
    EntityType,
    GraphPredicate,
    MemoryRecord,
    MemoryType,
    Sensitivity,
    SourceKind,
    TrustContext,
)

Memory types:

MemoryType.EPISODIC
MemoryType.WORKING
MemoryType.SEMANTIC
MemoryType.COMPETENCE
MemoryType.PLAN_GRAPH
MemoryType.ENTITY

Built-in entity ontology values are open-ended. Use EntityType.OTHER or any project-specific string when none of the built-ins fit.

Development

cd clients/python
pip install -e .[dev]
pytest

Requirements

• Python >= 3.10
• grpcio >= 1.80.0
• protobuf >= 6.31.1
• A running Membrane daemon, default localhost:9090