Skip to content

Recipe: Session Checkpointing

Goal

Keep a conversation alive across process restarts: checkpoint an AgentSession to a JSON file after every completed turn, resume it in another process, continue it under another model, and handle a restore that crosses providers. The concept page is History and Checkpointing.

Prerequisites

  • doc_processing_toolkit installed with at least one provider key in the environment.
  • A directory the process can write to (the examples use checkpoints/).

Checkpoint after every turn

from pathlib import Path

from doc_processing_toolkit import Agent

CHECKPOINT = Path("checkpoints/contract-review.json")

agent = Agent(model="claude-haiku-4-5", instructions="You review contracts. Be brief.")
session = agent.session()

for question in ("Summarise clause 4.", "Who bears the risk in clause 5?"):
    result = await session.run(question, files=["contract.pdf"] if question.startswith("Summarise") else None)
    session.to_json(path=CHECKPOINT)       # atomic: a crash mid-write keeps the previous file intact
    print(result.output)

The file after the loop holds two completed turns and the accounting for both. to_json returns the JSON text as well, for a host that stores envelopes in a database column instead of a file.

Resume in another process

from doc_processing_toolkit import Agent, AgentSession

agent = Agent(model="claude-haiku-4-5", instructions="You review contracts. Be brief.")
session = AgentSession.from_json("checkpoints/contract-review.json", agent=agent)

print(session.restored_from_schema_version)   # 3
print(session.metadata["run_count"])          # 2
if session.prefix_changed_on_resume:
    print(session.prefix_change_report)       # the instructions or tools moved since the save

result = await session.run("List the parties.")
session.to_json(path="checkpoints/contract-review.json")

Resend a turn that did not complete

from doc_processing_toolkit import Agent, AgentSession

session = AgentSession.from_json("checkpoints/contract-review.json", agent=agent)

record = session.interrupted_turn            # None unless the last turn failed
if record is not None:
    print(record["error_type"], record["error"], record["failed_at"])
    result = await session.run(record["message"])   # the files are not kept: pass them again if there were any
    session.to_json(path="checkpoints/contract-review.json")

The failed turn's question is never left in history unanswered: it lives on the record until the next turn starts. Checkpoint inside your except block as well, so the record reaches the next process.

Know what you are about to write

report = session.persistence_report
print(report["size_bytes"], report["message_count"], report["media_count"])
for part in report["media"]:
    print(part["message_index"], part["kind"], part["mime_type"], part["size_bytes"])

A PDF or an image attached with files= is stored inline (base64) on Anthropic, OpenAI and Fireworks and as a Files API reference on Google, so the report is where a host learns that an envelope is about to be 30 MB.

Save without a store

The default. Nothing to configure; the envelope is self-contained.

from doc_processing_toolkit import Agent, AgentSession

agent = Agent(model="gemini-3.5-flash-lite", instructions="You describe media. Be brief.")
session = agent.session()
result = await session.run("What is in this clip?", files=["clip.wav"])
print(result.media_uploads)                        # [{"uri": ".../files/...", "expiration_time": "...", ...}]
session.to_json(path="checkpoints/clip.json")      # "media_policy": "inline"; the clip's bytes are in the "media" table

restored = AgentSession.from_json("checkpoints/clip.json", agent=agent)
result = await restored.run("And how long is it?")  # after 48 h the clip is re-uploaded from the table first
print(restored.media_report)                       # [] — or [{"reason": "expired_reuploaded", ...}]

On Anthropic, OpenAI and Fireworks the attachment is inline base64 in messages; on Google a Files API upload is a file_uri in messages plus its bytes in the media table, kept up to media_max_inline_bytes (8 MiB) per part. Above that, the save logs a WARNING and media_report carries over_inline_threshold: configure a store, or pass media_max_inline_bytes=None.

Save with a store

from doc_processing_toolkit import Agent, AgentSession

agent = Agent(
    model="claude-haiku-4-5",
    instructions="You review contracts. Be brief.",
    media_store="checkpoints/media",                     # a DirectoryMediaStore; "auto" now means externalize
)
session = agent.session()
await session.run("Summarise clause 4.", files=["contract.pdf"])
session.to_json(path="checkpoints/contract-review.json")
# the PDF is checkpoints/media/<sha256>.pdf; the envelope holds "$media:<sha256>.pdf" in its place

# The same process or another one: the agent's store is used.
restored = AgentSession.from_json("checkpoints/contract-review.json", agent=agent)

# An agent configured without a store names it on the restore instead.
plain = Agent(model="claude-haiku-4-5", instructions="You review contracts. Be brief.")
restored = AgentSession.from_json("checkpoints/contract-review.json", agent=plain, media_store="checkpoints/media")

# What the restore could not re-inline (empty when every blob was found).
for entry in restored.media_report:
    print(entry["reason"], entry.get("mime_type"), entry.get("size"))   # e.g. missing_blob application/pdf 48231

The same policies per save, without configuring the agent:

session.to_json(path="checkpoints/review.json", media="externalize", media_store="checkpoints/media")
session.to_json(path="checkpoints/review-small.json", media="strip")     # placeholders, no bytes
session.to_json(path="checkpoints/review-full.json", media="inline")     # everything, even with a store configured

A blob deleted from the store does not fail the restore: the part becomes a text placeholder and media_report records missing_blob. Choose between the two setups with the guidance in Setting up a store; an object-storage store is the same two-method protocol against your own client.

Continue under another model of the same provider

from doc_processing_toolkit import Agent, AgentSession

envelope = session.to_dict()                                 # saved under claude-haiku-4-5
bigger = Agent(model="claude-opus-5-5", instructions="You review contracts. Be brief.")
session = AgentSession.from_dict(envelope, agent=bigger)     # quiet: the history reaches the new model as it is

A restore that crosses providers

from doc_processing_toolkit import Agent, AgentSession, SessionCompatibilityError

try:
    session = AgentSession.from_json("checkpoints/contract-review.json", agent=Agent(model="gpt-6-luna"))
except SessionCompatibilityError as exc:
    # Only a history that cannot be converted: it ends inside a tool round, a call has no name,
    # a result has no call. Truncate before that turn under the original provider and retry.
    print(exc)
else:
    print(session.converted_from)                       # "anthropic"
    for entry in session.conversion_report:             # what did not cross, and every re-keyed id
        print(entry["kind"], entry.get("artefact"), entry["index"], entry.get("detail"))
    result = await session.run("And clause 5?")         # continues on gpt-6-luna with the tool rounds intact
    session.to_json(path="checkpoints/contract-review.json")   # the report rides the envelope

The PDF attached on the Anthropic turn crosses as an input_file; a thinking block does not, and the report says so. A history of plain text turns (no tool rounds, no attachments, no retained reasoning) is provider-neutral, converts to itself and reports nothing.

Continue a live session on another model, in place

session = Agent(model="gpt-6-luna", instructions="You review contracts. Be brief.").session()
await session.run("Summarise clause 4.", files=["contract.pdf"])

session.rebind(Agent(model="claude-haiku-4-5", instructions="You review contracts. Be brief."))
print(session.converted_from, session.metadata["served_models"][-1])   # "openai", {"event": "rebind", ...}
await session.run("And clause 5?")                                      # runs on Anthropic

Same provider, the history is untouched; another provider, it is converted with the same report as a restore. A running turn refuses the rebind with RuntimeError.

What to validate

  • The checkpoint file is rewritten after every completed turn and never contains a half-turn.
  • from_json on the file gives run_count equal to the number of turns saved.
  • A restore under another model of the same provider logs nothing at WARNING or above.
  • A restore of a tool-using history under another provider converts it: converted_from names the source, conversion_report lists every drop and re-key, and the next turn runs with the tool rounds intact. Only a history that ends inside a tool round raises SessionCompatibilityError.
  • With media_store configured, the envelope holds no base64 payload ($media: markers instead) and from_json restores the original parts byte for byte; restored.media_report is empty.
  • On Google, result.media_uploads names every upload of the turn and the envelope's media table carries its bytes (or store reference) and expiration_time.

Test sources

  • tests/unit/test_session_envelope_v3.py: the version-3 keys, to_json / from_json, persistence_report, the refusal per dialect and served_models.
  • tests/unit/test_session_envelope_version.py: versioning and the prefix fingerprint.
  • tests/unit/test_media_policy.py: the four policies per dialect, DirectoryMediaStore, the Google media table and the dead-URI check, restored-history validation, preserve_media.
  • tests/unit/test_history_conversion.py: the four readers and writers, all twelve directions, the report, the refusals, from_dict conversion, rebind, the kill switch.