CrewAI Persistence — SQL¶

A CrewAI FlowPersistence backend for any SQLAlchemy database (SQLite, PostgreSQL, MySQL, …) with built-in message pruning. It persists flow state between steps so your flows can resume from any saved checkpoint, and it can automatically cap your message history before each write — no changes to your flow code or state model required.

Current version

crewai-persistence-sql 0.1.0 · Requires Python >=3.10,<3.14 (CrewAI constraint), crewai>=1.0.0, and SQLAlchemy>=2.0

What it is¶

SQLFlowPersistence subclasses crewai.flow.persistence.base.FlowPersistence (a Pydantic BaseModel + ABC in CrewAI 1.x) and stores CrewAI flow state in a single relational table via SQLAlchemy Core. Configuration (url, table_name, messages_key) is declared as Pydantic fields, while the runtime objects (the SQLAlchemy engine and the reducer) are held as PrivateAttr. One provider works for any SQLAlchemy-supported database. It accepts an optional MessageReducer that prunes the message list at the persistence layer.

Persistence is keyed by the flow state's id field — the flow_uuid. One row is stored per flow run: each save upserts the latest state (an UPDATE, falling back to INSERT when no row matched), so load_state returns the most recent. Flow state is serialised to a JSON string in a data text column. save_state accepts either a Pydantic BaseModel or a plain dict; load_state returns a dict (or None).

Installation¶

BaseWith pruning

pip install crewai-persistence-sql

pip install "crewai-persistence-sql[reducer]"

The [reducer] extra pulls in agentstate-reducer, required only if you pass a reducer. Database drivers ship as extras too: [postgres] (psycopg 3) and [mysql] (PyMySQL). SQLite needs no extra driver.

Database and table setup¶

Behaviour	Detail
Engine	Built from `url` via `create_engine`, or supplied directly as `engine=`
Table	Created automatically by `init_db()` via `metadata.create_all` if absent
Schema	Two columns — `flow_uuid` (`String(255)`, primary key) and `data` (`Text`, not null)

You must provide either url or engine; if both are omitted a ValueError is raised. The database itself (the PostgreSQL/MySQL server and target database) must already exist — SQLAlchemy creates the table, not the database.

Connection URLs¶

The url is a standard SQLAlchemy connection string. Provide credentials inside the URL, or build the engine yourself and pass engine=.

SQLite (dev)PostgreSQL (prod)MySQLPre-built engine

export DB_URL="sqlite:///flows.db"

export DB_URL="postgresql+psycopg://<user>:<password>@<host>:5432/<database>"

export DB_URL="mysql+pymysql://<user>:<password>@<host>:3306/<database>"

Build the engine yourself (custom pool size, TLS, timeouts) and pass it as engine=; the url is then ignored.

from sqlalchemy import create_engine
engine = create_engine("postgresql+psycopg://...", pool_size=10)

Quick start¶

Normal flow¶

import os
from crewai.flow.flow import Flow, start, listen
from crewai.flow.persistence import persist
from crewai_persistence_sql import SQLFlowPersistence

persistence = SQLFlowPersistence(
    url=os.environ.get("DB_URL", "sqlite:///flows.db"),
    table_name="crewai_flow_states",
)

@persist(persistence)
class MyFlow(Flow):
    @start()
    def first_step(self):
        return {"status": "started", "value": 42}

    @listen(first_step)
    def second_step(self, data):
        return data

flow = MyFlow()
flow.kickoff()

Conversational flow (with message history and pruning)¶

import os
from crewai.flow.flow import Flow, start, listen
from crewai.flow.persistence import persist
from agentstate_reducer import MessageReducer
from agentstate_reducer.models import ReducerConfig
from crewai_persistence_sql import SQLFlowPersistence

reducer = MessageReducer(config=ReducerConfig(min_messages=10, max_messages=20))

persistence = SQLFlowPersistence(
    url=os.environ.get("DB_URL", "sqlite:///flows.db"),
    table_name="crewai_flow_states",
    reducer=reducer,         # prune before each save
    messages_key="messages", # state key holding the message list (default)
)

@persist(persistence)
class ChatFlow(Flow):
    @start()
    def handle_turn(self):
        # messages accumulate here; pruning happens automatically at save time
        messages = self.state.get("messages", [])
        messages.append({"role": "human", "content": "Tell me about SQL databases."})
        # ... call your LLM here ...
        messages.append({"role": "ai", "content": "A SQL database stores data in relational tables..."})
        return {"messages": messages}

flow = ChatFlow()
flow.kickoff()

The @persist decorator

@persist is imported from crewai.flow.persistence. Applied to a Flow subclass with your persistence instance, it transparently calls save_state / load_state keyed by the flow state's id (the flow_uuid).

API reference¶

`SQLFlowPersistence(url=None, *, table_name="crewai_flow_states", engine=None, reducer=None, messages_key="messages")`¶

Parameter	Type	Default	Description
`url`	`str \\| None`	`None`	SQLAlchemy connection URL; required unless `engine` is given
`table_name`	`str`	`"crewai_flow_states"`	Table name (created if absent)
`engine`	`Engine \\| None`	`None`	Pre-built SQLAlchemy engine; if provided, `url` is ignored
`reducer`	`MessageReducer \\| None`	`None`	Optional pruner — see Built-in message pruning
`messages_key`	`str`	`"messages"`	State key that holds the message list

Provide url or engine

If neither url nor engine is supplied, init_db() raises ValueError("Provide either 'url' or 'engine'").

Methods¶

Method	Description
`init_db()`	Build the engine (from `url`) and create the table if absent (called automatically by `__init__`)
`save_state(flow_uuid, method_name, state_data)`	Persist flow state (UPDATE then INSERT fallback, keyed by `flow_uuid`); accepts a `BaseModel` or dict
`load_state(flow_uuid)`	Load the most recently saved state as a dict; returns `None` if not found

Built-in message pruning¶

Long-running conversational flows accumulate message history with every turn, inflating row size, increasing storage costs, and eventually blowing past LLM context limits.

Pass a MessageReducer and the backend prunes the message list inside save_state() before the row is written to the database. Your flow code and state model stay untouched.

When len(messages) > max_messages, the oldest human/ai messages are removed until min_messages remain. System-prompt index 0, system/function messages, and tool messages (unless their parent ai message is pruned) are preserved.

Full reducer configuration

For preserve_first, cascade_tool_messages, summarize_fn, token budgeting, and role aliases, see the reducer overview and token budget docs.

Data model¶

Each call to save_state upserts a single row keyed by flow_uuid. Only the latest state for each flow run is stored (an UPDATE on the matching row, or an INSERT when none exists).

Column	Type	Description
`flow_uuid`	`String(255)` (primary key)	Unique identifier for the flow run
`data`	`Text` (not null)	The full state dict serialised as a JSON string

JSON serialisation

The state dict is serialised into the data column as a JSON string; load_state parses it back into a dict. The method_name argument is accepted by save_state but not persisted — the table intentionally keeps a minimal two-column schema portable across SQLite, PostgreSQL, and MySQL.