Storage Backends

By default, Agentix persists sessions to disk under ~/.agentix/projects. For distributed or cloud deployments, plug in a custom storage backend.

File-based (default)

from agentix import AgentixAgentOptions

options = AgentixAgentOptions(
    persistence_enabled=True,
    storage_base_path="~/.agentix/projects",   # default
)

Redis backend

Requires pip install agentix[redis].

from agentix import AgentixAgentOptions, RedisStorageBackend

storage = RedisStorageBackend(
    url="redis://redis.internal:6379/0",
    prefix="myapp",              # namespace for all Redis keys (default: "agentix")
    decode_responses=True,       # default: True
)

options = AgentixAgentOptions(
    storage_backend=storage,
    persistence_enabled=True,
)

Redis key layout:

• {prefix}:session:{session_id} — session metadata hash
• {prefix}:messages:{session_id} — Redis list of JSON messages
• {prefix}:index:{agent_id} — sorted set of session IDs (scored by updated_at)

Custom backend

Subclass StorageBackend to plug in any storage system (PostgreSQL, S3, DynamoDB, etc.):

from agentix import StorageBackend, AgentixAgentOptions

class MyPostgresBackend(StorageBackend):

    async def acreate_session(
        self, agent_id: str, metadata: dict | None = None
    ) -> str:
        """Create a new session and return its ID."""
        ...

    async def aget_session(self, session_id: str) -> dict | None:
        """Return session metadata dict, or None if not found."""
        ...

    async def alist_sessions(
        self, agent_id: str, cwd: str | None = None
    ) -> list[dict]:
        """Return sessions for the agent, newest first."""
        ...

    async def afork_session(self, session_id: str) -> str:
        """Copy session and all messages to a new ID; return the new ID."""
        ...

    async def adelete_session(self, session_id: str) -> None:
        """Delete session and all its messages."""
        ...

    async def asave_message(
        self, session_id: str, role: str, content: str | list | dict
    ) -> None:
        """Append a single message to the session."""
        ...

    async def aload_messages(self, session_id: str) -> list[dict]:
        """Return all messages for the session in chronological order."""
        ...

    def close(self) -> None:
        """Release any resources (connections, pools, etc.)."""
        ...

options = AgentixAgentOptions(
    storage_backend=MyPostgresBackend(),
    persistence_enabled=True,
)

Session metadata shape

{
    "session_id": str,
    "agent_id": str,
    "created_at": "2024-01-15T14:30:00Z",   # ISO 8601
    "updated_at": "2024-01-15T14:31:00Z",
    "metadata": dict | None,
}

Message shape

{
    "role": "user" | "assistant" | "system",
    "content": str | list | dict,
    "timestamp": "2024-01-15T14:30:45Z",     # ISO 8601
}

Sync variants

StorageBackend also exposes synchronous methods (create_session, get_session, etc.) that default to running the async version in a thread via asyncio.to_thread. Override them only when a native synchronous implementation is more efficient.

Disabling persistence

options = AgentixAgentOptions(
    persistence_enabled=False,   # sessions exist only in memory
)

Sessions still persist in memory for the lifetime of the AgentixClient instance. Session history is lost when the client is destroyed.