Memsy
Memsy

API Reference

MemsyControlClient

Control-plane client for billing, API keys, usage, and account management.

MemsyControlClient and AsyncMemsyControlClient wrap the Memsy control-plane API (api/). This is a separate service from the hot-path memory engine — it handles account management, billing, API key lifecycle, and usage reporting.

from memsy import MemsyControlClient, AsyncMemsyControlClient

When to use this client

Use MemsyControlClient when you need to:

  • Look up account or org identity (me())
  • Read usage metrics and billing summaries
  • Programmatically create or rotate API keys (admin-only)
  • Browse ingested events in the console
  • Express interest in the Pro plan

For memory operations (ingest, search, clear), use MemsyClient instead.

Constructor

MemsyControlClient(
    base_url: str,
    api_key: str,
    timeout: float = 30.0,
    max_retries: int = 3,
    retry_backoff: float = 1.0,
)
ParameterTypeDefaultDescription
base_urlstr—Control-plane base URL, e.g. "https://api.memsy.io/api".
api_keystr—API key (msy_...). Sent as Authorization: Bearer <key>.
timeoutfloat30.0Request timeout in seconds.
max_retriesint3Max automatic retries on HTTP 429.
retry_backofffloat1.0Base exponential backoff in seconds.

Context manager

import os

with MemsyControlClient(
    base_url=os.environ["MEMSY_CONTROL_URL"],
    api_key=os.environ["MEMSY_API_KEY"],
) as control:
    me = control.me()

Top-level methods

me()

Return identity information for the authenticated API key.

me() -> MeResponse
me = control.me()
print(me.email, me.tier, me.org_role)

health()

Check if the control-plane is healthy.

health() -> HealthResponse

Sub-resource accessors

control.usage

Usage metrics for the current org.

# Summary for the current billing period
summary: UsageSummaryResponse = control.usage.summary()

# Timeseries (daily by default)
ts: UsageTimeseriesResponse = control.usage.timeseries(
    dimension="events_ingested",   # optional filter
    granularity="daily",           # "daily" | "hourly"
    period_start="2026-04-01",     # optional ISO-8601 date
    period_end="2026-04-30",
)

control.billing

Billing summary and invoice history.

summary: BillingSummary = control.billing.summary()
invoices: list[Invoice] = control.billing.invoices()

Note

Billing endpoints require billing_enabled=True on your org. If billing is not yet enabled, these calls raise BillingNotEnabledError. Use control.interest.express(...) to join the waitlist.

control.keys

API key management. Requires org_role == "org:admin" — plain API keys will receive AuthorizationError.

# List all keys (includes quota info)
key_list: ApiKeyListResponse = control.keys.list()

# Create a new key
new_key: CreateKeyResponse = control.keys.create(
    name="ci-deploy",
    scopes=["read", "write"],
    expires_at="2027-01-01T00:00:00Z",  # optional
)
print(new_key.raw_key)  # shown once — store immediately

# Per-key usage stats (returns list[dict] — raw usage records)
usage: list[dict] = control.keys.usage(key_id)

# Revoke a key
control.keys.delete(key_id)

control.events

Browse raw ingested events for your org (console view).

events: EventListResponse = control.events.list(
    actor_id="user_42",            # optional filter
    session_id="session_1",        # optional filter
    kind="user_message",           # optional filter
    sort="ts_desc",                # "ts_desc" | "ts_asc"
    limit=50,
    offset=0,
)

control.interest

Express interest in the Pro plan or check waitlist status.

# Express interest
resp: ProInterestResponse = control.interest.express(
    email="you@example.com",
    name="Your Name",
    company="Acme Inc",        # optional
    use_case="AI agent memory", # optional
)

# Check whether your org has already expressed interest
already_expressed: bool = control.interest.status()

control.connectors

Connect knowledge sources and choose what they sync. Providers today: slack, google_drive, s3, notion, github, onedrive.

Org-scoped providers require an admin. slack, s3, notion and github are a single shared connection per org — only an org admin, or an 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 read-only access. google_drive and onedrive are user-scoped: each member connects their own account, and an admin may audit but never mutate it. requires_org_admin(provider) is a client-side pre-flight helper only — the server enforces the rule.

from memsy import MemsyControlClient, ResourceSelection

# 1. Start OAuth — send the end user to authorize_url
connection: ConnectorConnection = control.connectors.create("slack")

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

# 3. Select what to sync — activates the connector and starts the backfill.
#    Replaces the previous selection, so always send the full set.
connector: Connector = control.connectors.configure_resources(
    connection.connector_id,
    [ResourceSelection.from_item(r) for r in resources],
)

control.connectors.status("slack")        # member-safe: {connected, display_name}
control.connectors.list_providers()
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 methods:

  • S3 does not use OAuth — create("s3") 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.
  • GitHub — list_resources() returns repos; list_branches(connector_id, repo=...) expands one repo's branches lazily.
  • OneDrive — list_resources(connector_id, parent_id=...) drills into a folder.
  • Google Drive — cannot enumerate server-side; list_resources() returns only the already-selected files. Real selection happens in the browser Google Picker, configured via picker_config(connector_id) (owner-only, mints a live Google token).

Async variant

AsyncMemsyControlClient is the async counterpart. Every method is identical but returns a coroutine.

import os

async with AsyncMemsyControlClient(
    base_url=os.environ["MEMSY_CONTROL_URL"],
    api_key=os.environ["MEMSY_API_KEY"],
) as control:
    me = await control.me()
    summary = await control.usage.summary()
    events = await control.events.list(limit=20)

Sub-resources (control.usage, control.billing, control.keys, control.events, control.interest, control.connectors) are all available with await equivalents.

See also

  • MemsyClient — hot-path memory client.
  • Models — all control-plane response dataclasses.
  • Exceptions — BillingNotEnabledError, KeyLimitReachedError, SeatLimitReachedError, and others raised by this client.

Async variant

AsyncMemsyControlClient is the async counterpart. Every method is identical but returns a coroutine.

import os

async with AsyncMemsyControlClient(
    base_url=os.environ["MEMSY_CONTROL_URL"],
    api_key=os.environ["MEMSY_API_KEY"],
) as control:
    me = await control.me()
    summary = await control.usage.summary()
    events = await control.events.list(limit=20)

Sub-resources (control.usage, control.billing, control.keys, control.events, control.interest) are all available with await equivalents.

See also

  • MemsyClient — hot-path memory client.
  • Models — all control-plane response dataclasses.
  • Exceptions — BillingNotEnabledError, KeyLimitReachedError, SeatLimitReachedError, and others raised by this client.