MongoDB State Store

Flexible document storage for agent state with TTL and rich metadata

MongoDB stores agent state as flexible documents — ideal for nested metadata and schema-less data.

python

from praisonaiagents import Agent, MemoryConfig
from praisonaiagents import db

agent = Agent(
    name="DocumentBot",
    instructions="You are a helpful assistant.",
    memory=MemoryConfig(
        db=db(state_url="mongodb://localhost:27017/praisonai"),
        session_id="mongo-session",
    ))
agent.start("Store my preferences in MongoDB state")

The user sets preferences; MongoDB stores flexible agent state as documents.

Diagram

Quick Start

Simple Usage

bash

pip install pymongo praisonai

python

from praisonaiagents import Agent, MemoryConfig
from praisonaiagents import db

agent = Agent(
    name="DocumentBot",
    memory=MemoryConfig(
        db=db(state_url="mongodb://localhost:27017/praisonai"),
        session_id="session-1",
    ))
agent.start("Hello!")

With Configuration

Use MongoDBStateStore directly for collection and database control:

python

from praisonai.persistence import create_state_store

store = create_state_store(
    "mongodb",
    url="mongodb://localhost:27017",
    database="praisonai",
    collection="agent_state",
)
store.set("user:123:prefs", {"theme": "dark"}, ttl=3600)

How It Works

MongoDB is a state store — it holds key-value agent state, not full conversation history. Pair it with a SQL conversation backend when you need both.

Diagram

Feature	Description
Document storage	Nested objects and arrays without a fixed schema
TTL index	Automatic expiry via `expires_at` field
Hash operations	`hget`, `hset`, `hgetall` for structured state

Configuration Options

Option	Type	Default	Description
`url`	`str`	`"mongodb://localhost:27017"`	MongoDB connection URL
`database`	`str`	`"praisonai"`	Database name
`collection`	`str`	`"state"`	Collection for state documents

URL formats

python

db(state_url="mongodb://localhost:27017/praisonai")
db(state_url="mongodb://user:pass@host:27017/praisonai?replicaSet=rs0")

For async workloads, use create_state_store("async_mongodb", ...) — see Async MongoDB (motor).

Tuning the connection

The sync mongodb backend sets sensible pymongo defaults for pooling, timeouts, and retries. Pass any pymongo URI option in the connection string’s query to override them for your deployment. The async async_mongodb backend passes the URL to motor as-is, so only motor’s own defaults apply there.

URI option	SDK default	Why it’s set
`maxPoolSize`	`50`	Bounded client-side connection pool
`minPoolSize`	`5`	Keeps a few warm connections ready
`maxIdleTimeMS`	`30000`	Closes connections idle for 30s
`serverSelectionTimeoutMS`	`5000`	5s before giving up on finding a server
`retryWrites`	`True`	Retries one eligible write on a retryable error
`retryReads`	`True`	Retries one eligible read on a retryable error

python

# Standalone MongoDB without a replica set — turn off retryable writes
db(state_url="mongodb://localhost:27017/praisonai?retryWrites=false")

# Small worker — smaller pool and faster failure when MongoDB is down
db(state_url="mongodb://localhost:27017/praisonai?maxPoolSize=10&serverSelectionTimeoutMS=3000")

# Production cluster — replica set + credentials, keep defaults
db(state_url="mongodb://user:pass@host1:27017,host2:27017/praisonai?replicaSet=rs0")

Diagram

Method reference

Both mongodb and async_mongodb implement the full StateStore contract. Every method has an async_* twin that runs natively on the event loop; the sync method is a wrapper safe to call from any context.

Method	Async twin	Purpose
`get(key)`	`async_get(key)`	Fetch a stored value (respects TTL)
`set(key, value, ttl=None)`	`async_set(key, value, ttl=None)`	Store a value with optional TTL in seconds
`delete(key)`	`async_delete(key)`	Remove a key
`exists(key)`	`async_exists(key)`	Check if a key exists
`list_keys(prefix=None)`	`async_list_keys(prefix=None)`	List keys with an optional prefix filter
`clear(prefix=None)`	`async_clear(prefix=None)`	Delete many keys, optionally under a prefix
`keys(pattern="*")`	`async_keys(pattern="*")`	Glob pattern match — `*` and `?`; patterns are `re.escape`-d before wildcards re-enable
`ttl(key)`	`async_ttl(key)`	Remaining seconds until expiry, or `None` if no TTL / expired
`expire(key, ttl)`	`async_expire(key, ttl)`	Set or reset TTL on an existing key
`hget(key, field)`	`async_hget(key, field)`	Read one field from a hash-shaped value
`hset(key, field, value)`	`async_hset(key, field, value)`	Write one field into a hash-shaped value (upsert)
`hgetall(key)`	`async_hgetall(key)`	Read all fields of a hash-shaped value
`hdel(key, *fields)`	`async_hdel(key, *fields)`	Delete fields from a hash; returns count actually removed

Async MongoDB (motor)

The async backend (async_mongodb) uses motor for non-blocking I/O — pick it inside FastAPI, async handlers, or any live event loop.

bash

pip install motor

Sync or async?

Diagram

Agent with an async store

Create the async store, then pass it to the agent’s memory config:

python

from praisonaiagents import Agent, MemoryConfig
from praisonai.persistence import create_state_store

# Aliases "motor" and "mongodb_async" resolve to the same store
store = create_state_store("async_mongodb", url="mongodb://localhost:27017", database="praisonai")

agent = Agent(
    name="AsyncBot",
    instructions="You are a helpful assistant.",
    memory=MemoryConfig(db=store, session_id="async-session"),
)

await agent.astart("Remember I prefer dark mode")

Direct store usage

Every operation has a native async_* method for use inside an event loop:

python

import asyncio
from praisonai.persistence import create_state_store

async def main():
    store = create_state_store("async_mongodb", url="mongodb://localhost:27017")

    # Set / get with TTL (seconds)
    await store.async_set("user:123:prefs", {"theme": "dark"}, ttl=3600)
    prefs = await store.async_get("user:123:prefs")

    # Pattern matching (glob: * and ?)
    session_keys = await store.async_keys("session:*")

    # TTL introspection and refresh
    remaining = await store.async_ttl("user:123:prefs")   # seconds left, or None
    await store.async_expire("user:123:prefs", 7200)       # reset to 2h

    # Hash operations for structured state
    await store.async_hset("user:123", "email", "a@b.com")
    email = await store.async_hget("user:123", "email")
    all_fields = await store.async_hgetall("user:123")
    removed = await store.async_hdel("user:123", "email", "phone")  # count removed

    await store.async_close()

asyncio.run(main())

Request flow

A typical FastAPI request reads and writes session state through the async store without blocking the loop.

Diagram

Best Practices

Anchor key patterns with a fixed prefix

Call keys("session:*") with a known prefix rather than passing raw user input into the glob, so a listing returns only the keys you intend to match.

Pair with a conversation backend

Use database_url for chat history and state_url for fast agent state — MongoDB handles state only.

Set TTL for ephemeral data

Pass ttl on set() for session-scoped preferences that should expire automatically.

Use replica sets in production

Append replicaSet= to the URL for high availability.

Choose collection names per app

Set collection="prod_state" vs collection="staging_state" to isolate environments on one cluster.

Redis State Store

In-memory state for sub-millisecond access

Database Persistence

Overview of conversation and state backends

MongoDB State Store

Quick Start

How It Works

Configuration Options

URL formats

Tuning the connection

Method reference

Async MongoDB (motor)

Sync or async?

Agent with an async store

Direct store usage

Request flow

Best Practices

Related