TanStack AI Persistence, Resumable Streams & Memory with Redis

TanStack AI defines the contracts for an agent's production state — chat persistence, resumable streaming, locks, memory — and ships in-memory implementations that only work inside one process. @upstash/agentkit-tanstack-ai implements them on Upstash Redis, so they hold across serverless instances, page reloads, and devices.

Import	Plugs into	Feature
`upstashPersistence`	`withPersistence()`, `withGenerationPersistence()`	Transcripts, runs, human-in-the-loop interrupts, metadata, generation jobs, and generated files.
`upstashStream`	`durability` on the response	Resume a stream after a reload, or open the same thread on another device.
`upstashLocks`	`withLocks()`	Distributed locks for TanStack AI middleware, such as its sandbox setup.
`upstashMemory`	`memoryMiddleware()`	Long-term memory ranked in Redis Search.
`toolCache`, `rateLimit`	`middleware`	Skip repeated tool calls; throttle users before the model runs.
`createSearchTools`	`tools`	`search` / `aggregate` / `count` over your own documents (RAG).

npm install @upstash/agentkit-tanstack-ai @tanstack/ai

Note

AgentKit reads UPSTASH_REDIS_REST_URL / UPSTASH_REDIS_REST_TOKEN from the environment by default. Pass a redis client to any helper to use a different one.

How to persist TanStack AI chats in Redis#

Persistence plugs into TanStack AI's withPersistence() middleware, which comes from its persistence package:

npm install @tanstack/ai-persistence

import { chat } from "@tanstack/ai";import { withPersistence } from "@tanstack/ai-persistence";import { upstashPersistence } from "@upstash/agentkit-tanstack-ai/persistence";const persistence = upstashPersistence();chat({ adapter, messages, threadId, middleware: [withPersistence(persistence)] });

This covers every TanStack AI persistence store: messages, runs, interrupts, and metadata for chats, plus generationRuns and artifacts for one-shot generation jobs such as images or speech. The blobs store for generated bytes is added when you pass an Upstash Blob bucket:

import { Bucket } from "@upstash/blob";const persistence = upstashPersistence({ bucket: Bucket.fromEnv() });

Reads UPSTASH_REDIS_REST_URL and UPSTASH_REDIS_REST_TOKEN from the environment. Bucket.fromEnv() reads UPSTASH_BLOB_TOKEN from the environment, which is only needed when you store generated files.

Runs are indexed by thread, so reconnecting to a live run (findActiveRun) is a single index read. Each write is one command or one Lua script, so concurrent instances cannot interleave it.

Options

upstashPersistence({  redis, // optional: defaults to Redis.fromEnv()  prefix: "agentkit:tanstack", // optional: base key prefix  messagesTtlSeconds: 60 * 60 * 24 * 30, // optional: expire idle transcripts (default: never)  bucket: Bucket.fromEnv(), // optional: Upstash Blob bucket for generated files});

Pass bucket to also store the bytes of generated files (images, audio, video) in Upstash Blob. Without it, there is no blobs store.

How to resume a TanStack AI stream after a reload#

import { chat, toServerSentEventsResponse } from "@tanstack/ai";import { upstashStream } from "@upstash/agentkit-tanstack-ai";export async function POST(request: Request) {  const stream = chat({ adapter, messages, threadId });  return toServerSentEventsResponse(stream, { durability: { adapter: upstashStream(request) } });}

Every chunk is written to a Redis Stream before it is sent. A client that reconnects with Last-Event-ID (or ?offset) replays what it missed and keeps following the live run, whichever instance serves the request. Without a Request, use upstashStream({ runId, offset }).

Reads UPSTASH_REDIS_REST_URL and UPSTASH_REDIS_REST_TOKEN from the environment.

Options

upstashStream(request, {  ttlSeconds: 86_400, // optional: how long a run stays resumable  pollIntervalMs: 150, // optional: how often a caught-up reader checks for new chunks  firstChunkDeadlineMs: 2_000, // optional: how long a join waits for a run that has not started});

How to use distributed locks with TanStack AI#

import { withLocks } from "@tanstack/ai/locks";import { upstashLocks } from "@upstash/agentkit-tanstack-ai";chat({ adapter, messages, middleware: [withLocks(upstashLocks()), withSandbox(sandbox)] });

withLocks doesn't lock anything by itself. It gives the lock store to later middleware, which lock the one step they must not run twice: withSandbox uses it so two concurrent requests for a thread don't both create a sandbox, and your own middleware can use it through getLocks(ctx). It does not serialize whole chat turns. Unlike TanStack's InMemoryLockStore, which only works inside one process, upstashLocks() coordinates across instances.

Each lock is a lease that is renewed while the critical section runs. If the lease is lost, the section's signal aborts.

Reads UPSTASH_REDIS_REST_URL and UPSTASH_REDIS_REST_TOKEN from the environment.

Options

upstashLocks({  leaseMs: 30_000, // optional: lease lifetime without renewal  acquireTimeoutMs: 30_000, // optional: how long to wait for a held key  retryDelayMs: 100, // optional: delay between attempts});

How to add long-term memory to TanStack AI#

Memory plugs into TanStack AI's memoryMiddleware(), which comes from its memory package:

npm install @tanstack/ai-memory

import { memoryMiddleware } from "@tanstack/ai-memory";import { upstashMemory } from "@upstash/agentkit-tanstack-ai/memory";chat({  adapter,  messages,  middleware: [    memoryMiddleware({      adapter: upstashMemory(),      // derive these server-side from the session, never from the request body      scope: (ctx) => ({ threadId: ctx.threadId, userId: session.userId }),    }),  ],});

Before each turn, the most relevant memories for the user's message are added to the system prompt, labelled by where they came from. The model gets a save_memory tool for durable facts, and each turn's user message is captured too. Memory is per user across threads by default.

Reads UPSTASH_REDIS_REST_URL and UPSTASH_REDIS_REST_TOKEN from the environment.

Options

upstashMemory({  topK: 5, // optional: memories injected per turn  scopeBy: "user", // optional: "user" (across threads) or "thread"  captureUserMessages: true, // optional: store each turn's user message  saveTool: true, // optional: offer the save_memory tool  waitForIndexing: true, // optional: make a save recallable on the very next turn});

How to cache tools and rate limit with TanStack AI#

import { rateLimit, Ratelimit, toolCache } from "@upstash/agentkit-tanstack-ai";chat({  adapter,  messages,  tools: [getWeather, sendEmail],  middleware: [    rateLimit({ limiter: Ratelimit.slidingWindow(10, "60 s"), identifier: userId }),    toolCache({ tools: ["get_weather"], userId, ttlSeconds: 600 }),  ],});

toolCache only caches the tools you list — list deterministic, side-effect-free tools only. rateLimit fails the run with RateLimitExceededError before the model is called. For an HTTP 429 instead, call createRateLimit({ limiter }).limit(userId) in your route before chat().

Both middlewares and createRateLimit read UPSTASH_REDIS_REST_URL and UPSTASH_REDIS_REST_TOKEN from the environment.

How to add RAG with TanStack AI#

import { s } from "@upstash/redis";import { createSearchTools } from "@upstash/agentkit-tanstack-ai";const tools = createSearchTools({  indexName: "products",  schema: s.object({ name: s.string(), price: s.number(), category: s.string().noTokenize() }),});chat({ adapter, messages, tools });

The tool descriptions are generated from the schema, and the index is created on first use.

Reads UPSTASH_REDIS_REST_URL and UPSTASH_REDIS_REST_TOKEN from the environment.

Telemetry#

AgentKit adds its package name and version as a header on your Redis client's requests. To turn it off, set UPSTASH_DISABLE_TELEMETRY, or pass enableTelemetry: false to a helper.

AgentKit on GitHub

Source and README for the package.

TanStack AI

The framework these backends plug into.