memsy

Official Python SDK for Memsy — persistent memory for AI agents and applications.

📚 Docs: docs.memsy.io — guides, API reference, and migration notes.

Installation

pip install memsy

Quick Start

import os
from memsy import MemsyClient, EventPayload

client = MemsyClient(
    base_url=os.environ["MEMSY_BASE_URL"],
    api_key=os.environ["MEMSY_API_KEY"],
)

# Remember something
client.ingest([EventPayload(
    actor_id="user_1", session_id="session_1",
    kind="user_message", content="I prefer dark mode in all apps",
    role_id="role_eng",       # optional: scope to a role
    team_id="team_platform",  # optional: scope to a team
)])

# Recall it later
results = client.search("user preferences", actor_id="user_1")
for r in results.results:
    print(r.content)

That's it. The client handles connection pooling, retries, and response parsing automatically.

---

Configuration

Authentication

The SDK uses Bearer token authentication. Pass your API key when creating the client:

client = MemsyClient(
    base_url=os.environ["MEMSY_BASE_URL"],
    api_key=os.environ["MEMSY_API_KEY"],
)

Retry Configuration

Configure retry behavior for rate-limited requests:

client = MemsyClient(
    base_url=os.environ["MEMSY_BASE_URL"],
    api_key=os.environ["MEMSY_API_KEY"],
    max_retries=3,         # default: 3
    retry_backoff=1.0,     # default: 1.0 seconds
    timeout=30.0,          # default: 30.0 seconds
)

---

API Reference — MemsyClient (hot path)

ingest(events)

Store a batch of events. Events are processed asynchronously into long-term memories. Organization context is inferred from the API key — do not pass org_id.

result = client.ingest([
    EventPayload(
        actor_id="user_1", session_id="s1",
        kind="user_message", content="...",
        role_id="engineer",  # optional — used by hierarchical promotion
        team_id="platform",  # optional — used by hierarchical promotion
    ),
    EventPayload(
        actor_id="user_1", session_id="s1",
        kind="assistant_message", content="...",
    ),
])
print(result.event_ids)  # ['01J...', '01J...']

EventPayload fields

Field	Type	Required	Description
actor_id	str	Yes	End-user or agent identifier
session_id	str	Yes	Conversation/session identifier
kind	str	Yes	user_message, assistant_message, tool_result, app_event
content	str	Yes	Text content of the event
ts	str	No	ISO 8601 timestamp (server uses now() if omitted)
metadata	str	No	JSON-serialised string for custom attributes
role_id	str	No	Scope this event to a specific role in the hierarchy
team_id	str	No	Scope this event to a specific team in the hierarchy

search(query, *, actor_id, session_id, exclude_session_id, limit, threshold, include_source_events)

Retrieve relevant memories using natural language.

results = client.search(
    query="what does the user prefer?",
    actor_id="user_1",            # optional — restrict to one actor; omit for org-wide search
    session_id="sess_abc",        # optional — scope to one conversation
    limit=10,                     # default: 10
    threshold=0.0,                # minimum relevance score, default: 0.0 (no filter)
    include_source_events=True,   # attach source events to each result
)

for r in results.results:
    print(r.score, r.content)

    # Typed metadata properties (always safe — return None/[] if absent)
    print(r.title, r.summary, r.tags)
    print(r.strength, r.confidence)
    print(r.session_id)           # conversation it came from; None if it belongs to none

    # Typed source events when include_source_events=True
    for evt in r.source_events:
        print(evt.event_id, evt.kind, evt.content)

Scope a search to one conversation with session_id, or to everything except one with exclude_session_id. Send neither — the default — to search across all of them. The two are mutually exclusive; sending both non-empty returns a 422. An empty string is treated as unset, so session_id="" alongside exclude_session_id is a plain exclude search rather than an error — pass None, never "". Memories belonging to no conversation (promoted role/team/org knowledge) are returned either way.

Searchable ids may contain only letters, digits and _ - : . @ /, up to 256 characters. Ingest checks the length but not the character set, so an id with a space, #, + or any non-ASCII character stores fine and returns a 422 when you search it. Slugify ids you don't control — anything derived from a chat title will hit this.

status(event_ids)

Check if ingested events have been processed into memories.

status = client.status(event_ids=result.event_ids)
print(status.completed_ids)
print(status.pending_ids)

health()

h = client.health()
print(h.status)          # "ok"
print(h.version)         # "2.1.0"
print(h.billing_enabled) # True | False | None
print(h.components)      # {"async_memsy": "ok", "sync_memsy": "ok", ...}

---

Onboarding Hierarchy

MemsyClient exposes three sub-resource accessors for managing the org → role → team hierarchy that scopes memory promotion. Deleting a record removes only the customization — memories with that org_id / role_id / team_id are unaffected.

client.orgs

# Create / get / update
org = client.orgs.create(org_id="my-org", name="My Org", focus="AI assistant context")
org = client.orgs.get("my-org")
org = client.orgs.update("my-org", focus="Updated focus")

# Regenerate the LLM-written promotion_prompt
org = client.orgs.regenerate_prompt("my-org")

# List all visible orgs
orgs = client.orgs.list()

# Delete customization record
client.orgs.delete("my-org")

client.roles

role = client.roles.create(org_id="my-org", name="Engineering", focus="Software engineers")
roles = client.roles.list(org_id="my-org")
role = client.roles.get(role_id="role-id", org_id="my-org")
role = client.roles.update("role-id", "my-org", name="Senior Engineering")
role = client.roles.regenerate_prompt("role-id", "my-org")
client.roles.delete("role-id", "my-org")

client.teams

team = client.teams.create(org_id="my-org", name="Platform", focus="Infrastructure team")
teams = client.teams.list(org_id="my-org")
team = client.teams.get(team_id="team-id", org_id="my-org")
team = client.teams.update("team-id", "my-org", focus="Platform & infra")
team = client.teams.regenerate_prompt("team-id", "my-org")
client.teams.delete("team-id", "my-org")

---

Console Memories

Browse memories stored for the authenticated org via client.memories:

# Paginated list with filters
page = client.memories.list(
    kind="semantic",         # semantic | episodic | procedural
    type="preference",       # fact | preference | norm | decision | ...
    sort="observed_at_desc", # default
    limit=50,
    offset=0,
)
print(page.total, len(page.items))

# Aggregate statistics
stats = client.memories.stats()
print(stats.total_memories, stats.avg_confidence)
print(stats.by_type)    # {"fact": 12, "preference": 5, ...}

# Retrieve a single memory by UUID
item = client.memories.get("550e8400-e29b-41d4-a716-446655440000")
print(item.text, item.strength, item.confidence)

---

Actors

List the actors that have memories in an org via client.actors. Read-only: actors are derived from the org's memories on every read rather than stored, so there is nothing to create, rename or delete.

page = client.actors.list(
    sort="first_memory_desc",   # first_memory_desc | first_memory_asc
                                # last_memory_desc  | actor_id_asc
    q="slack",                  # substring match on actor_id, case-insensitive
    limit=100,
    offset=0,
)
print(page.total)               # distinct actors in the org

for actor in page.items:
    print(actor.actor_id, actor.first_memory_at, actor.memory_count)

# One actor; raises MemsyAPIError (404) if it has no memories in this org
actor = client.actors.get("slack:T0B7FKNKKTR:U0B7TN132E9")

org_id defaults to the authenticated caller's org on both methods.

Read the timestamps carefully. They are named for the memories, not the actor, because that is what they measure — there is no actor creation event to read:

• first_memory_at is min(created_at). It moves forward if the oldest memory is reaped or decayed away, which a real creation timestamp never could.
• last_memory_at is max(created_at). It is not a "last used" signal: it only advances when a new memory is written. Reads never touch it.

Check page.truncated on large orgs. When set, the server hit its row-scan cap before reading every memory. The actors listed are real, but the list may be incomplete and every memory_count is an under-count — treat the counts as unusable while it is set. This is also why there is no sort by memory count.

One human can legitimately appear as several actors: a connector-minted slack:T123:U456 and a plugin-derived hash are the same person on two surfaces.

---

Control-Plane Client (MemsyControlClient)

The control-plane is a separate API service that manages account settings, billing, API keys, usage reporting, and raw event browsing. Use a second client pointed at the control-plane URL:

from memsy import MemsyControlClient

control = MemsyControlClient(
    base_url=os.environ["MEMSY_CONTROL_URL"],  # e.g. https://api.memsy.io/api
    api_key=os.environ["MEMSY_API_KEY"],
)

control.me()

Returns identity information for the authenticated caller:

me = control.me()
print(me.email, me.tier, me.org_id, me.is_billing_admin)

control.events.list()

Browse raw ingested events (requires assigned seat):

events = control.events.list(
    actor_id="user_1",      # optional filter
    session_id="sess_abc",  # optional filter
    kind="user_message",    # optional filter
    sort="ts_desc",         # default
    limit=50,
)
for e in events.items:
    print(e.ts, e.actor_id, e.content)

control.usage (admin-only)

summary = control.usage.summary()
print(summary.tier, summary.period_start, summary.period_end)
for dim in summary.dimensions:
    print(f"{dim.dimension}: {dim.used} / {dim.limit}")

ts = control.usage.timeseries(dimension="api_calls", granularity="daily")
for point in ts.data:
    print(point.date, point.quantity)

control.billing (admin-only)

billing = control.billing.summary()
print(billing.tier, billing.purchased_seats, billing.subscription_status)
if billing.payment_method:
    print(billing.payment_method.brand, billing.payment_method.last4)

invoices = control.billing.invoices()
for inv in invoices:
    print(inv.status, inv.amount_due, inv.currency)

control.keys (admin-only)

# List existing keys
resp = control.keys.list()
print(resp.active_count, resp.max_keys)
for key in resp.keys:
    print(key.key_id, key.prefix, key.scopes, key.is_active)

# Create a new key — raw_key is returned ONCE, store it securely
new_key = control.keys.create("ci-pipeline", scopes=["read"])
print(new_key.raw_key)

# Per-key usage records
records = control.keys.usage(new_key.key_id)

# Delete a key by ID
control.keys.delete(new_key.key_id)

control.connectors

Connect knowledge sources — Slack, Google Drive, S3, Notion, GitHub, OneDrive — and choose what they sync.

Who may connect what. Org-scoped providers (slack, s3, notion, github) are one shared connection per org: only an org admin — or any API key, which the server treats as a service caller acting org-wide — may connect, configure, sync or disconnect them. A seated non-admin member gets AuthorizationError (403) and may only read. User-scoped providers (google_drive, onedrive) are per-member: the owner manages their own connection, and an admin can see it for auditing but never mutate it. requires_org_admin(provider) answers this client-side for pre-flight checks — the server is the enforcement point.

from memsy import MemsyControlClient, ResourceSelection, requires_org_admin

control = MemsyControlClient(base_url=MEMSY_CONTROL_URL, api_key=MEMSY_API_KEY)

control.connectors.list_providers()
# ['slack', 'google_drive', 's3', 'notion', 'github', 'onedrive']
requires_org_admin("slack")         # True — admin/API key only
requires_org_admin("google_drive")  # False — per-member connection

# 1. Start OAuth and send the end user to the consent screen
connection = control.connectors.create("slack")
print(connection.authorize_url)

# 2. The provider redirects back to Memsy's own callback, which attaches the
#    token. list_resources() answers 409 until then, so poll:
resources = control.connectors.wait_until_authorized(connection.connector_id)

# 3. Pick what to sync — this activates the connector and starts the backfill.
#    The selection replaces the previous one, so send the full set.
control.connectors.configure_resources(
    connection.connector_id,
    [ResourceSelection.from_item(r) for r in resources if r.name != "random"],
)

# Later
control.connectors.status("slack")            # member-safe: connected? (no ids)
control.connectors.list()                     # admin/API-key view
control.connectors.get(connector_id)          # status, last_sync_at, last_error
control.connectors.sync(connector_id)         # manual refresh (must be active)
control.connectors.delete(connector_id)       # disconnect + revoke

Provider-specific bits:

Provider	Notes
slack	Org-scoped. resource_type="channel".
s3	Org-scoped, no OAuth — create() returns 400. Use configure_s3(access_key_id=..., secret_access_key=..., region=..., bucket=...), which validates the credentials, selects the bucket and starts the backfill in one call.
notion	Org-scoped. resource_type="workspace".
github	Org-scoped. list_resources() returns repos (resource_type="repo"); call list_branches(connector_id, repo=...) to expand one repo's branches lazily.
google_drive	User-scoped. Drive can't enumerate server-side, so list_resources() returns only what's already selected — real selection happens in the browser Google Picker via picker_config(connector_id), then the chosen file ids go to configure_resources().
onedrive	User-scoped. list_resources(connector_id, parent_id=...) drills into folders.

Gmail is deprecated and is no longer a registered provider.

control.interest

# Express Pro interest
control.interest.express(
    email="you@company.com", name="Your Name",
    company="Acme", use_case="AI assistant memory"
)

# Check status
already_expressed = control.interest.status()  # bool

---

Usage Tracking

The SDK automatically parses usage and rate limit headers from every response:

result = client.search("test query")

if result.usage:
    print(f"Plan: {result.usage.plan}")
    print(f"API calls: {result.usage.api_calls} / {result.usage.api_calls_limit}")

if result.rate_limit:
    print(f"Rate limit remaining: {result.rate_limit.remaining}")

---

Error Handling

from memsy.exceptions import (
    AuthenticationError,
    AuthorizationError,
    BillingNotEnabledError,
    FeatureNotAvailable,
    KeyLimitReachedError,
    OrgIdNotAllowedError,
    OrgLimitReachedError,
    RateLimitExceeded,
    SeatLimitReachedError,
    SeatRequiredError,
    UsageLimitExceeded,
    MemsyConnectionError,
    MemsyAPIError,
)

try:
    results = client.search("preferences")
except AuthenticationError:
    print("Invalid API key")
except AuthorizationError as e:
    print(f"Missing required scope: {e.required_scope}")
except SeatRequiredError:
    print("This endpoint requires an assigned seat")
except FeatureNotAvailable as e:
    print(f"Feature '{e.feature}' not available on {e.current_tier}")
    print(f"Upgrade at: {e.upgrade_url}")
except OrgIdNotAllowedError:
    print("Free-tier orgs cannot pass org_id in request bodies")
except OrgLimitReachedError as e:
    print(f"Org limit: {e.current}/{e.limit}")
except KeyLimitReachedError as e:
    print(f"API key limit: {e.current}/{e.limit}")
except BillingNotEnabledError as e:
    print(f"Express interest at: {e.interest_path}")
except SeatLimitReachedError as e:
    print(f"Seats: {e.assigned_seats} assigned / {e.purchased_seats} purchased")
except RateLimitExceeded as e:
    print(f"Rate limited — retry after {e.retry_after}s")
except UsageLimitExceeded as e:
    print(f"Quota exceeded for {e.dimension}: {e.current}/{e.limit}")
except MemsyConnectionError:
    print("Could not reach Memsy")
except MemsyAPIError as e:
    print(f"API error {e.status_code}: {e.detail}")

Exception Hierarchy

MemsyError
├── MemsyConnectionError      # Network/timeout errors
└── MemsyAPIError             # Non-2xx responses
    ├── AuthenticationError   # 401 — Invalid/missing API key
    ├── AuthorizationError    # 403 — Wrong scope or admin-required
    ├── FeatureNotAvailable   # 403 — Feature gated by tier
    ├── OrgIdNotAllowedError  # 400 — org_id sent on free tier
    ├── SeatRequiredError     # 403 — Endpoint needs an assigned seat
    ├── OrgLimitReachedError  # 403 — Tier org cap hit
    ├── KeyLimitReachedError  # 403 — Tier API key cap hit
    ├── BillingNotEnabledError# 403 — Billing endpoint on free tier
    ├── SeatLimitReachedError # 409 — Seat purchase limit reached
    ├── RateLimitExceeded     # 429 — Rate limit hit
    └── UsageLimitExceeded    # 429 — Quota exceeded

---

Auto-Retry

The SDK automatically retries on 429 (rate limit) responses with exponential backoff:

• Default: 3 retries with 1.0s base backoff
• Respects Retry-After header if present
• After max retries, raises RateLimitExceeded

---

Async Usage

All clients have async equivalents: AsyncMemsyClient and AsyncMemsyControlClient.

import asyncio
from memsy import AsyncMemsyClient, AsyncMemsyControlClient, EventPayload

async def main():
    async with AsyncMemsyClient(base_url="https://...", api_key="msy_...") as client:
        await client.ingest([EventPayload(
            actor_id="user_1", session_id="s1",
            kind="user_message", content="I prefer dark mode",
            role_id="role_eng",
        )])
        results = await client.search("user preferences")

        # Sub-resources are also async
        orgs = await client.orgs.list()
        stats = await client.memories.stats()

    async with AsyncMemsyControlClient(base_url="https://.../api", api_key="msy_...") as control:
        me = await control.me()
        events = await control.events.list()

asyncio.run(main())

---

Context Manager (auto-close)

with MemsyClient(base_url="...", api_key="msy_...") as client:
    results = client.search("recent topics")

---

Migration Guide

Upgrading from 0.2.x to 0.3.0

org_id removed

The deprecated org_id parameter is gone. Any code passing it will raise TypeError.

# Before (0.2.x — deprecated but accepted)
results = client.search("query", org_id="org_1")
client.ingest([EventPayload(org_id="org_1", actor_id="u1", ...)])

# After (0.3.0)
results = client.search("query")
client.ingest([EventPayload(actor_id="u1", ...)])

MemsyAuthError removed

Replace with AuthenticationError:

# Before
from memsy import MemsyAuthError

# After
from memsy import AuthenticationError

New role/team scoping on EventPayload

# New in 0.3.0 — optional, ignored if not set
EventPayload(
    actor_id="u1", session_id="s1", kind="user_message", content="...",
    role_id="role_eng",       # optional
    team_id="team_platform",  # optional
)

Upgrading from 0.1.x to 0.2.x

See the 0.2.0 CHANGELOG entry for details on the x-api-key → Authorization: Bearer header change and the initial org_id deprecation.

---

Publishing (maintainers)

cd python-sdk
pip install hatch
hatch build
hatch publish