Skip to main content

checkpoints.middleware

Auto-checkpoint middleware.

Hooks into turn/tool events to automatically create checkpoints based on the configured frequency.

AutoCheckpointMiddleware Objects​

class AutoCheckpointMiddleware()

Automatically create conversation checkpoints.

Integrates with the agent run loop to save checkpoints at configurable intervals (every tool call, every turn, or manual-only).

Parameters​

config : CheckpointConfig Checkpoint configuration. store : CheckpointStore | None Pre-built store; if None one is created from config.

on_turn_start​

async def on_turn_start(
        messages: list[dict[str, Any]],
        metadata: dict[str, Any] | None = None
) -> ConversationCheckpoint | None

Called at the start of each model request turn.

Creates a checkpoint if frequency is every_turn.

on_tool_call​

async def on_tool_call(
        tool_name: str,
        messages: list[dict[str, Any]],
        metadata: dict[str, Any] | None = None
) -> ConversationCheckpoint | None

Called before each tool execution.

Creates a checkpoint if frequency is every_tool.

save_manual​

async def save_manual(
        label: str,
        messages: list[dict[str, Any]],
        metadata: dict[str, Any] | None = None) -> ConversationCheckpoint

Manually save a named checkpoint (used by the agent tool).

list_checkpoints​

async def list_checkpoints() -> list[ConversationCheckpoint]

List all available checkpoints.

get_checkpoint​

async def get_checkpoint(checkpoint_id: str) -> ConversationCheckpoint | None

Retrieve a specific checkpoint.

from_spec​

@classmethod
def from_spec(cls,
              spec_checkpoints: dict | None) -> "AutoCheckpointMiddleware"

Create from AgentSpec.