3tears-langgraph
Three-tier LangGraph checkpoint saver: L1 (SQLite) -> L2 (NATS KV) -> L3 (PostgreSQL).
L1 and L2 are optional cache layers that degrade gracefully on failure. L3 (PostgreSQL) is the source of truth, reached through the AsyncQueryExecutor protocol so the same saver serves trusted services (direct asyncpg pool) and sandboxed agents (NATS L3 proxy).
L3 failures reach the caller, with one channel-scoped carve-out: aput_writes is called from LangGraph's executor teardown, where raising kills a turn that has already answered, so a failed crash-recovery write is logged and degraded. Writes on LangGraph's control channels (__interrupt__, __resume__, __error__, __scheduled__) still raise: losing one changes what the run does, and a lost __interrupt__ would silently skip a human-approval gate. A control-channel write also overwrites an earlier write on the same channel, matching the reference saver, so a second Command(resume=...) replaces the first rather than being dropped.
Installation
pip install 3tears-langgraph
Usage
from threetears.langgraph import (
AsyncpgPoolAdapter,
CheckpointScope,
ThreeTierCheckpointSaver,
)
# Trusted service with direct asyncpg.Pool: wrap once
saver = ThreeTierCheckpointSaver(
executor=AsyncpgPoolAdapter(pool),
scope=CheckpointScope.for_customer(customer_id),
)
# Sandboxed agent: NatsProxyL3Backend already implements
# AsyncQueryExecutor, pass it straight through
saver = ThreeTierCheckpointSaver(
executor=nats_l3_backend,
scope=CheckpointScope.for_customer(customer_id),
)
graph = builder.compile(checkpointer=saver)
Scope is required
scope has no default. A saver either names the customer whose checkpoints it addresses, or says in writing that it deliberately names none:
saver = ThreeTierCheckpointSaver(
executor=AsyncpgPoolAdapter(pool),
scope=CheckpointScope.unscoped(reason="single-tenant deployment"),
)
CheckpointScope.for_customer(...) folds the customer into the stored thread_id, and therefore into the L3 bound parameter, the L2 bucket key, and the L1 thread key — a saver scoped to one customer cannot name another customer's row at any tier. It also unlocks adelete_customer_threads(), the whole-tenant purge, which refuses on an unscoped saver.
CheckpointScope.unscoped(...) is a legitimate answer, not a placeholder: it produces byte-identical keys and statements to a pre-tenancy saver, so an existing deployment adopts the required parameter by adding this one argument and migrating no data. The reason is mandatory, logged at WARNING on construction, and greppable in source, so "which deployments still run unscoped, and why" has an answer.
Adopting a real customer later is a data change rather than a code change: existing rows live under a bare thread id and a scoped saver will not find them, so they must be re-keyed (UPDATE checkpoints SET thread_id = $customer || '/' || thread_id, likewise checkpoint_writes, plus L2 invalidation). No re-key script ships here and none can — which customer owns which thread lives in the host's own tables, which this library has never seen.
Middleware
The package ships platform-level AgentMiddleware for langchain.agents.create_agent — the framework-aligned successor to the old hand-rolled AgentNodeHook / ToolNodeHook protocols. Consumer-specific policy lives in each consumer as its own middleware; only the reusable platform seams live here:
PromptCachingMiddleware(wrap_model_call) — annotates a leading bare-string system message with Anthropiccache_control={"type": "ephemeral"}when the model supports it, then normalizes cache-hit/creation counters ontousage_metadata["cache_usage"]. Non-Anthropic adapters degrade silently to bare-string system messages.ToolResultOffloadMiddleware(wrap_tool_call) — when aToolResultOffloaderis injected onconfig["configurable"]and a tool result exceedsoffload_threshold_chars, stores the full content out-of-band and shows the model"<summary>\n\n[ctx:<handle>]"(the structuredartifactis preserved). Opt-in: no offloader ⇒ byte-for-byte no-op.ObjectCatalogMiddleware(wrap_tool_call) — when a tool returns anObjectHandlein its result artifact and anObjectCatalogeris injected, persists a catalog record under the verified call identity. Soft-fail side-effect: a catalog error never breaks the tool result.
from langchain.agents import create_agent
from threetears.langgraph import (
ObjectCatalogMiddleware,
PromptCachingMiddleware,
ToolResultOffloadMiddleware,
)
agent = create_agent(
model=chat_anthropic,
tools=tools,
middleware=[
PromptCachingMiddleware(),
ToolResultOffloadMiddleware(),
ObjectCatalogMiddleware(),
],
)
# after a run, PromptCachingMiddleware has stamped:
# message.usage_metadata["cache_usage"]
# == {"cache_read_input_tokens": ..., "cache_creation_input_tokens": ..., "cached_tokens": ...}
The offload / catalog contracts (ToolResultOffloader, ObjectCataloger) are pure structural Protocols exported from threetears.langgraph.offload / threetears.langgraph.catalog; a consumer injects a concrete implementation on config["configurable"] (e.g. tool_result_offloader, object_cataloger) without the package taking any dependency on the consumer's context store.
See 3tears/docs/prompt-caching.md for the full caching contract, summarization interaction, downstream wiring checklist, and a worked example.
Streaming
The package ships StreamingResponse, a transport-agnostic primitive that owns the lifecycle of one streaming response: start -> any number of emit_token / emit_tool_call_* -> mutually-exclusive end (success) or error (failure) terminal. run_graph(compiled_graph, state, config) consumes a LangGraph astream_events(version="v2") loop with the start/end ordering managed; on graph exception it fires error(code="AGENT_FAILED", ...) and re-raises so the caller still sees the failure on the synchronous path.
The wire vocabulary is fixed: StreamStartEvent / StreamTokenEvent / StreamEndEvent / StreamErrorEvent / ToolCallStartEvent / ToolCallEndEvent / ToolCallProgressEvent, dispatched via the StreamEvent discriminated union and the parse_stream_event(payload) adapter. The transport seam is the StreamTransport Protocol -- one method, async def publish(self, payload: bytes) -> None. Any wire (NATS subject, websocket, chunked HTTP body) satisfies it.
from threetears.langgraph import StreamingResponse, StreamTransport
class WebSocketStreamTransport:
"""example transport for a websocket consumer."""
def __init__(self, ws): self._ws = ws
async def publish(self, payload: bytes) -> None:
await self._ws.send_bytes(payload)
stream = StreamingResponse(
transport=WebSocketStreamTransport(ws),
correlation_id=correlation_id,
conversation_id=conversation_id,
start_time_monotonic=request_start,
)
final_state = await stream.run_graph(compiled_graph, state, config)
A reference adapter can bind the primitive to a per-correlation-id stream subject via nc.publish_raw. Tool-call observation envelopes flow through ToolCallProgressHook reading the active StreamingResponse from config["configurable"]["streaming_response"].
1 maintainer
Release files for 3tears-langgraph 0.46.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| 3tears_langgraph-0.46.0.tar.gz | 163.7 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| 3tears_langgraph-0.46.0-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 274.2 kB
Release files / 3tears_langgraph-0.46.0.tar.gz
| Download URL | 3tears_langgraph-0.46.0.tar.gz |
|---|---|
| Size | 163.7 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
a6e92898e63554c7d7f7a18601bac15fc6244d3d4468c54a381b23c0de400d6f
|
|
BLAKE2b-256 checksum How to use checksums |
941a56d708b932f42c8c8dfefa3bf6175c660c05220b7913a6e608c45e58ddcd
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Sep 17, 2026.
Transparency logRelease files / 3tears_langgraph-0.46.0-py3-none-any.whl
| Download URL | 3tears_langgraph-0.46.0-py3-none-any.whl |
|---|---|
| Size | 110.5 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
dbca3464838b19addf8da655e586202aeaa61e5044552e0c4d65521b0f2064a5
|
|
BLAKE2b-256 checksum How to use checksums |
98212b76070bd9fece9d3169965060eb65d65d953ae6e230889e2a3b37a7ee8e
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Sep 17, 2026.
Transparency log
