feat: HTTP-proxy LangGraph checkpointer #258
Open
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
danielmillerp
force-pushed
the
dm/checkpointer
branch
from
b1675a5 to
0c56de4
Compare
Replace direct Postgres checkpointing with HTTP-proxied checkpoint operations through the agentex backend API. Agents no longer need DATABASE_URL or direct DB connections for LangGraph state persistence. - Add HttpCheckpointSaver that proxies through AsyncAgentex client - Add create_checkpointer() factory using the HTTP checkpointer - Replace langgraph-checkpoint-postgres dep with langgraph-checkpoint - Export checkpointer module from adk package Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
danielmillerp
force-pushed
the
dm/checkpointer
branch
from
0c56de4 to
db0f5ab
Compare
3 tasks
RoxyFarhad
reviewed
Feb 9, 2026
|
|
||
| async def _post(self, path: str, body: dict[str, Any]) -> Any: | ||
| """POST JSON to the backend and return parsed response.""" | ||
| resp = self._client._client.post( # noqa: SLF001 |
Collaborator
There was a problem hiding this comment.
what is the double _client call here ? can we remove that
RoxyFarhad
reviewed
Feb 9, 2026
| ) | ||
| response = await resp | ||
| response.raise_for_status() | ||
| if response.status_code == 204: |
Collaborator
There was a problem hiding this comment.
why do we have this if statement
RoxyFarhad
reviewed
Feb 9, 2026
|
|
||
| @override | ||
| async def aget_tuple(self, config: RunnableConfig) -> CheckpointTuple | None: | ||
| configurable = config["configurable"] # type: ignore[reportTypedDictNotRequiredAccess] |
Collaborator
There was a problem hiding this comment.
do we wanna make these [""] type safe to be get() so it doesn't panic in the code? not sure if this would be caught anywhere but just thought id call out
RoxyFarhad
reviewed
Feb 9, 2026
| def get_tuple(self, config: RunnableConfig) -> CheckpointTuple | None: | ||
| import asyncio | ||
|
|
||
| return asyncio.get_event_loop().run_until_complete(self.aget_tuple(config)) |
Collaborator
There was a problem hiding this comment.
is the event loop always meant to exist? should we wrap these in a try block?
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
What this does
This PR replaces LangGraph's direct-Postgres checkpointer with an HTTP-proxy checkpointer that routes all checkpoint operations through the agentex backend API.
Why
LangGraph agents need to persist state (checkpoints) between messages. The built-in approach (
AsyncPostgresSaver) has each agent pod open its own Postgres connection pool. This doesn't scale — more agent pods means more connections, and we'd hit limits as we grow LangGraph usage. This is the same problem we already solved for Temporal: agents shouldn't talk to the DB directly. Instead, they go through the backend API, which manages a shared connection pool.How it works
We implemented LangGraph's
BaseCheckpointSaverabstract class — the same interface thatAsyncPostgresSaverimplements — but instead of running SQL queries, each method makes an HTTP POST to the backend:BaseCheckpointSavermethodaget_tuple()POST /checkpoints/get-tupleaput()POST /checkpoints/putaput_writes()POST /checkpoints/put-writesalist()POST /checkpoints/listadelete_thread()POST /checkpoints/delete-threadThe
HttpCheckpointSaveruses the existingAsyncAgentexhttpx client (withEnvAuthfor the agent API key), so authentication works the same way as every other SDK→backend call.Serialization stays in the SDK — complex Python objects are serialized via LangGraph's
serde, then base64-encoded for JSON transport. The backend just stores and retrieves the raw data.What changed
HttpCheckpointSaverclass — the HTTP-proxy checkpointer implementationcreate_checkpointer()factory — returns anHttpCheckpointSaverwired up with the SDK clientlanggraph-checkpoint-postgresandpsycopg[binary]dependencies (no longer needed)langgraph-checkpoint(base types only —BaseCheckpointSaver,CheckpointTuple, etc.)Impact on agents
Zero code changes. The
create_checkpointer()API is unchanged — agents just call it and get back a checkpointer. The only difference is they no longer needDATABASE_URLin their environment.Companion PR
/checkpointsendpoints, ORM models, migration, repository with 19 integration testsTest plan
🤖 Generated with Claude Code