# Saving User message manually for AsyncPostgreSaver checkpointer **URL:** \ **Category:** LangGraph\ **Tags:** python-help\ **Created:** [October 4, 2025, 12:08pm UTC](https://forum.langchain.com/t/saving-user-message-manually-for-asyncpostgresaver-checkpointer/1741 "2025-10-04T12:08:47Z")\ **Posts on this page:** 3\ **Page:** 1
**Author:** ![J3zus](https://yyz1.discourse-cdn.com/flex007/user_avatar/forum.langchain.com/j3zus/32/2064_2.png) [@J3zus](https://forum.langchain.com/u/J3zus)\ **Post date:** [October 4, 2025, 12:08pm UTC](https://forum.langchain.com/t/saving-user-message-manually-for-asyncpostgresaver-checkpointer/1741/1 "2025-10-04T12:08:47Z")
Hey I’m using AsyncPostgreSaver checkpointer and I want to save only new user message without invoking the graph. Checkpoints for this thread\_id already exist in checkpoints, checkpoint\_blobs, checkpoint\_writes tables in PostgreSQL. I just want to append a new message to last checkpoint with “aput” function. The code below doesn’t work. I’m getting checkpoint with no new message when executing aget\_tuple after saving a new checkpoint. How to do it correctly? Code: ```auto import asyncio import uuid from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver from langchain_core.messages import HumanMessage async def main(): config = {“configurable”: {“thread_id”: “28f6b69f-ad07-4937-9e65-fa6afa53f3f5”}} async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer: #await checkpointer.setup() state = await checkpointer.aget_tuple(config) checkpoint = state.checkpoint config = state.config print("state:", state) checkpoint["channel_values"]["messages"].append(HumanMessage(content="test", id=str(uuid.uuid4()))) await checkpointer.aput( config=config, checkpoint=checkpoint, metadata={}, new_versions={} ) state = await checkpointer.aget_tuple(config) print("state:", state) if __name__ ==" __main__": result = asyncio.run(main()) ``` ---
**Author:** ![pawel-twardziak](https://yyz1.discourse-cdn.com/flex007/user_avatar/forum.langchain.com/pawel-twardziak/32/960_2.png) [@pawel-twardziak](https://forum.langchain.com/u/pawel-twardziak)\ **Post date:** [October 4, 2025, 7:51pm UTC](https://forum.langchain.com/t/saving-user-message-manually-for-asyncpostgresaver-checkpointer/1741/2 "2025-10-04T19:51:51Z")
Hi @J3zus it is usually not a good practise to mutate `checkpoint` directly. I’d rather go with the dedicated API, such as `aput_writes` to append a message to the `messages` channel. Checkpointers persist a snapshot plus per-channel writes. Appending to a channel should be expressed as a write so it’s recorded in `checkpoint_writes` and merged by reducers. ```python async def main(): config = {"configurable": {"thread_id": "28f6b69f-ad07-4937-9e65-fa6afa53f3f5"}} async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer: # First-time only: await checkpointer.setup() state = await checkpointer.aget_tuple(config) if state is None: raise RuntimeError("No existing checkpoint for this thread_id") # Append a user message via per-channel writes await checkpointer.aput_writes( config=state.config, # should reference the target checkpoint_id writes=[("messages", [HumanMessage(content="test", id=str(uuid.uuid4()))])], task_id=str(uuid.uuid4()), task_path="", # keep it empty or "manual" to distinguish graph-external updates ) # Note: aget_tuple returns the snapshot. Per-channel writes are stored separately # and merged when creating a new checkpoint (e.g., on next graph step). updated = await checkpointer.aget_tuple(config) print("updated:", updated) ``` ### Why your snippet didn’t show the new message - Using `aput` with `new_versions={}` causes non-primitive channel values (like a list of messages) to be popped into `blob_values` but not persisted, so you read back a checkpoint without the new message. - Mutating `checkpoint["channel_values"]["messages"]` and passing that dict to `aput` does not register a channel write. The saver expects per-channel deltas in `writes` so they can be stored in `checkpoint_writes` and merged by reducers on read. ### References (official docs and source) - **How-to: Persistence** : [LangGraph docs](https://langchain-ai.github.io/langgraph/how-tos/persistence/) - **Reference: Checkpoints API** : [LangGraph docs](https://langchain-ai.github.io/langgraph/reference/checkpoints/) - **AsyncPostgresSaver.aput** : [docs](https://langchain-ai.github.io/langgraph/reference/checkpoints/#langgraph.checkpoint.postgres.aio.AsyncPostgresSaver.aput) - **AsyncPostgresSaver.aput\_writes** : [docs](https://langchain-ai.github.io/langgraph/reference/checkpoints/#langgraph.checkpoint.postgres.aio.AsyncPostgresSaver.aput_writes) - **Async Postgres saver (source)**: [aio.py in langgraph](https://github.com/langchain-ai/langgraph/blob/main/libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py) - **Package** : [langgraph-checkpoint-postgres on PyPI](https://pypi.org/project/langgraph-checkpoint-postgres/) ---
**Author:** ![skamboj](https://avatars.discourse-cdn.com/v4/letter/s/87869e/32.png) [@skamboj](https://forum.langchain.com/u/skamboj)\ **Post date:** [November 28, 2025, 10:39am UTC](https://forum.langchain.com/t/saving-user-message-manually-for-asyncpostgresaver-checkpointer/1741/3 "2025-11-28T10:39:11Z")
@pawel-twardziak I am trying something like: ``` latest_checkpoint = await checkpointer.aget_tuple(config) await checkpointer.adelete_thread(config["configurable"]["thread_id"]) if latest_checkpoint: LOGGER.info(f"Channel versions: {latest_checkpoint.checkpoint.get("channel_versions", {})}") await checkpointer.aput( latest_checkpoint.config, latest_checkpoint.checkpoint, latest_checkpoint.metadata, latest_checkpoint.checkpoint.get("channel_versions", {})) return response except Exception as e: LOGGER.error(f"Error invoking agent: {e}") raise e ``` but this is also not working, channel versions are not empty dict.