mnexium

Official Mnexium SDK for Python. Add persistent memory, conversation history, user profiles, and agent state to your AI apps.

Works with OpenAI, Anthropic, and Google Gemini. Bring your own API key.

Installation

pip install mnexium

Quick Start

import os
from mnexium import Mnexium, ProviderConfig

mnx = Mnexium(
    api_key="mnx_...",  # Optional - auto-provisions trial key if omitted
    openai=ProviderConfig(api_key=os.environ["OPENAI_API_KEY"]),
)

# Get a Subject handle (no network call)
alice = mnx.subject("user_123")
# Or auto-generate an ID:
anon = mnx.subject()

# Simple one-off message
response = alice.process("Hello!")
print(response.content)

# Multi-turn chat with history
from mnexium import ChatOptions

chat = alice.create_chat(ChatOptions(history=True, learn=True, recall=True))
chat.process("My favorite color is blue")
chat.process("What is my favorite color?")  # Remembers!

Core Concepts

Subject

A Subject is a logical identity (user, agent, org, device) that owns memory, profile, and state. Creating a Subject is instant — no network call.

alice = mnx.subject("user_123")

# Subject owns all resources
alice.memories.search("hobbies")
alice.memories.list()
alice.profile.get()
alice.state.get("preferences")
alice.claims.get("favorite_color")
alice.chats.list()

Chat

A Chat is a conversation thread with a stable chat_id. Chats belong to a Subject.

from mnexium import ChatOptions

chat = alice.create_chat(ChatOptions(
    model="gpt-4o-mini",
    history=True,    # Include previous messages
    learn=True,      # Extract memories from conversation
    recall=True,     # Inject relevant memories into context
    profile=True,    # Include user profile in context
))

chat.process("Hello!")
chat.process("What did I just say?")  # Has history

Streaming

Real-time streaming responses:

from mnexium import ChatProcessOptions

stream = chat.process(ChatProcessOptions(content="Tell me a story", stream=True))

for chunk in stream:
    print(chunk.content, end="", flush=True)

# Metadata available after stream completes
print(stream.total_content)  # Full accumulated text
print(stream.usage)          # Token counts
print(stream.chat_id)        # Chat ID

# Or collect the full response at once
stream2 = chat.process(ChatProcessOptions(content="Summarize", stream=True))
text = stream2.text()

Memory Policy Override

Override memory extraction policy per request:

from mnexium import ChatProcessOptions

# Use a specific memory policy ID
chat.process(ChatProcessOptions(
    content="Plan a trip for me",
    memory_policy="mpol_abc123",
))

# Disable memory policy for this request
chat.process(ChatProcessOptions(
    content="Do not learn from this",
    memory_policy=False,
))

For advanced/low-level calls, the SDK also sends header fallback:

• x-mnx-memory-policy: false when memory_policy=False
• x-mnx-memory-policy: <policy_id> when memory_policy is a string

Multi-Provider Support

The SDK auto-detects the provider from the model name:

mnx = Mnexium(
    openai=ProviderConfig(api_key=os.environ["OPENAI_API_KEY"]),
    anthropic=ProviderConfig(api_key=os.environ["ANTHROPIC_API_KEY"]),
    google=ProviderConfig(api_key=os.environ["GOOGLE_API_KEY"]),
)

alice = mnx.subject("user_123")
chat = alice.create_chat(ChatOptions(learn=True, recall=True))

# Switch models freely — memories are shared across providers
chat.process(ChatProcessOptions(content="I love hiking", model="gpt-4o-mini"))
chat.process(ChatProcessOptions(content="What do I love?", model="claude-sonnet-4-20250514"))
chat.process(ChatProcessOptions(content="Tell me my hobbies", model="gemini-2.0-flash"))

Records

Records are project-level structured entities (for example account, deal, ticket) with typed schemas, CRUD, filtering, and semantic search.

# Define a schema once
mnx.records.define_schema(
    "account",
    {
        "name": {"type": "string", "required": True},
        "industry": {"type": "string"},
        "arr": {"type": "number"},
    },
    display_name="Account",
    description="Business accounts",
)

# Insert structured records
record = mnx.records.insert("account", {"name": "TechCorp", "arr": 5000000})

API Reference

Memories

# Add a memory
mem = alice.memories.add("User prefers dark mode", source="settings")

# List memories
memories = alice.memories.list(limit=50)

# Semantic search
results = alice.memories.search("preferences", limit=5)

# Get a specific memory
memory = alice.memories.get("mem_abc123")

# Update a memory
alice.memories.update("mem_abc123", text="Updated text")

# Delete a memory
alice.memories.delete("mem_abc123")

# List superseded memories
old = alice.memories.superseded()

# Restore a superseded memory
alice.memories.restore("mem_abc123")

# Query recall events (which memories were used in a chat)
recalls = alice.memories.recalls(chat_id="chat_xyz")

Real-time Memory Events

Subscribe to memory changes via SSE:

events = alice.memories.subscribe()

for event in events:
    if event.type == "memory.created":
        print("New memory:", event.data)
    elif event.type == "memory.superseded":
        print("Memory replaced:", event.data)
    elif event.type == "memory.deleted":
        print("Memory removed:", event.data)

# Close the connection
events.close()

Event types: connected, memory.created, memory.updated, memory.deleted, memory.superseded, profile.updated, heartbeat

Claims (Structured Facts)

# Set a claim
alice.claims.set("favorite_color", "blue", confidence=0.95)

# Get a specific slot
color = alice.claims.get("favorite_color")

# List all claim slots
slots = alice.claims.list()

# Get current truth (all active values)
truth = alice.claims.truth()

# Get claim history
history = alice.claims.history()

# Retract a claim
alice.claims.retract("clm_abc123")

Profile

# Get profile
profile = alice.profile.get()

# Update profile fields
alice.profile.update([
    {"field_key": "display_name", "value": "Alice"},
    {"field_key": "timezone", "value": "America/New_York"},
])

# Delete a profile field
alice.profile.delete_field("timezone")

Agent State

Persist key-value state for workflows, wizards, and agent continuity:

# Set state (with optional TTL)
alice.state.set("current_task", {"step": 3}, ttl_seconds=3600)

# Get state
task = alice.state.get("current_task")

# Delete state
alice.state.delete("current_task")

Chat History

# List recent chats
from mnexium import ChatHistoryListOptions

chats = alice.chats.list(ChatHistoryListOptions(limit=10))

# Read messages from a specific chat
messages = alice.chats.read("chat_abc123")

# Delete a chat
alice.chats.delete("chat_abc123")

System Prompts

from mnexium import SystemPromptCreateOptions

# Create a prompt
prompt = mnx.prompts.create(SystemPromptCreateOptions(
    name="Customer Support",
    prompt_text="You are a helpful customer support agent...",
    is_default=True,
))

# List all prompts
prompts = mnx.prompts.list()

# Update a prompt
mnx.prompts.update(prompt["id"], prompt_text="Updated instructions...")

# Preview which prompts will be injected
resolved = mnx.prompts.resolve(subject_id="user_123")

# Use in chat
chat = alice.create_chat(ChatOptions(system_prompt=prompt["id"]))

# Delete a prompt
mnx.prompts.delete(prompt["id"])

Records

# Define or update a schema
mnx.records.define_schema(
    "deal",
    {
        "title": {"type": "string", "required": True},
        "value": {"type": "number"},
        "stage": {"type": "string"},
    },
)

# Get one schema / list all schemas
schema = mnx.records.get_schema("deal")
schemas = mnx.records.list_schemas()

# Insert
deal = mnx.records.insert("deal", {"title": "Enterprise Renewal", "value": 250000})

# Get by ID
fetched = mnx.records.get("deal", deal["record_id"])

# Partial update
updated = mnx.records.update("deal", deal["record_id"], {"stage": "closed_won"})

# Structured query
rows = mnx.records.query(
    "deal",
    where={"stage": "closed_won"},
    order_by="-value",
    limit=10,
)

# Semantic search
hits = mnx.records.search("deal", "enterprise renewal", limit=5)

# Delete
mnx.records.delete("deal", deal["record_id"])

Integrations

Manage inbound pull and webhook connectors that map external payloads into prompt-ready keys:

from mnexium import (
    IntegrationCreateOptions,
    IntegrationExecutionOptions,
    IntegrationOutputMapEntry,
    IntegrationWebhookOptions,
)

# Create a pull integration
integration = mnx.integrations.create(
    IntegrationCreateOptions(
        name="Weather Forecast",
        mode="pull",
        scope="project",
        endpoint_url="https://api.open-meteo.com/v1/forecast",
        method="GET",
        allow_live_fetch=True,
        cache_ttl_seconds=300,
        query_template={
            "latitude": "40.7128",
            "longitude": "-74.0060",
            "current": "temperature_2m,wind_speed_10m",
        },
        output_map=[
            IntegrationOutputMapEntry(key="weather_temp", path="current.temperature_2m"),
            IntegrationOutputMapEntry(key="weather_wind", path="current.wind_speed_10m"),
        ],
    )
)

# List / inspect / update / disable
integrations = mnx.integrations.list()
same_integration = mnx.integrations.get(integration["integration_id"])
mnx.integrations.update(integration["integration_id"], {"cache_ttl_seconds": 600})
mnx.integrations.delete(integration["integration_id"])

# Execute a dry run or refresh cache
preview = mnx.integrations.test(integration["integration_id"])
refreshed = mnx.integrations.sync(
    integration["integration_id"],
    IntegrationExecutionOptions(subject_id="user_123", chat_id="chat_456"),
)

# Send a signed webhook payload
mnx.integrations.webhook(
    integration["integration_id"],
    {"current": {"temperature_2m": 14.1, "wind_speed_10m": 13.8}},
    IntegrationWebhookOptions(secret="whsec_...", event_id="evt_123"),
)

print(preview["values"]["weather_temp"])
print(refreshed["cache_written"])

Configuration

from mnexium import Mnexium, ProviderConfig, MnexiumDefaults

mnx = Mnexium(
    api_key="mnx_...",                    # Optional — auto-provisions trial key
    base_url="https://mnexium.com/api/v1",  # API base URL
    timeout=30.0,                         # Request timeout (seconds)
    max_retries=2,                        # Retry count for failed requests
    openai=ProviderConfig(api_key="..."),    # OpenAI config
    anthropic=ProviderConfig(api_key="..."), # Anthropic config
    google=ProviderConfig(api_key="..."),    # Google config
    defaults=MnexiumDefaults(             # Default options for all process() calls
        model="gpt-4o-mini",
        learn=True,
        recall=False,
        profile=False,
        history=True,
    ),
)

Mnx Parameters

Every process() and chat.process() call supports these options:

Parameter	Type	Default	Description
model	str	"gpt-4o-mini"	Model to use
learn	bool/"force"	True	Extract memories from conversation
recall	bool	False	Inject relevant memories into context
profile	bool	False	Include user profile in context
history	bool	True	Prepend previous messages from this chat
log	bool	True	Save messages to chat history
summarize	bool/str	False	"light", "balanced", or "aggressive"
system_prompt	bool/str	True	True (auto), False (skip), or prompt ID
memory_policy	str/bool	—	Policy ID string, or False to disable policy for this request
records	object	—	Records controls: `learn: "force"
stream	bool	False	Enable streaming response
metadata	dict	—	Custom metadata attached to saved logs

Records extraction controls:

from mnexium.types import ProcessOptions, MnxRecordsConfig

response = mnx.process(
    ProcessOptions(
        content="Extract and save order details",
        records=MnxRecordsConfig(
            learn="force",         # "force" | "auto" | False
            tables=["orders"],     # Optional allowlist
            sync=True,             # Wait for write completion
            recall=True,           # Include prior records in context
        ),
    )
)

print(response.records)  # Sync extraction metadata (when returned by backend)

Trial Keys

If you don't provide an API key, Mnexium auto-provisions a trial key:

mnx = Mnexium(
    openai=ProviderConfig(api_key=os.environ["OPENAI_API_KEY"]),
)

alice = mnx.subject("user_123")
alice.process("Hello!")

# Get the provisioned key to save it
trial = mnx.get_trial_info()
if trial:
    print("Save this key:", trial["key"])

Claim your trial key at mnexium.com/claim.

Error Handling

from mnexium import Mnexium, AuthenticationError, RateLimitError, APIError

try:
    alice.process("Hello!")
except AuthenticationError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limited. Current: {e.current}, Limit: {e.limit}")
except APIError as e:
    print(f"API error {e.status}: {e}")

Raw Response

Access the full API response for advanced use:

response = chat.process("Hello!")
print(response.raw)       # Full API response dict
print(response.chat_id)   # Chat ID
print(response.usage)     # UsageInfo(prompt_tokens, completion_tokens, total_tokens)

Examples

See the examples/ directory for runnable demos:

Example	Description
hello_world.py	Hello world — chat, history, recall
streaming.py	Real-time streaming responses
events.py	Real-time memory event subscriptions
memories.py	Add, list, search, delete memories
claims.py	Claims extraction and manual setting
state.py	Agent state with TTL
profile.py	User profiles
prompts.py	System prompt management
multi_provider.py	Multi-provider (OpenAI, Claude, Gemini)
full_demo.py	Full API demo

Supported Models

• OpenAI: gpt-4o, gpt-4o-mini, gpt-4-turbo, gpt-4, gpt-3.5-turbo, o1, o1-mini, o3
• Anthropic: claude-3-opus, claude-3-sonnet, claude-3-haiku, claude-3-5-sonnet, claude-sonnet-4
• Google Gemini: gemini-2.0-flash-lite, gemini-2.5-flash, gemini-1.5-pro, gemini-1.5-flash

Links

• Documentation
• Blog
• Sign Up

License

MIT