> ## Documentation index > Fetch the complete documentation index at: https://upstash.com/docs/llms.txt > Ask it with GET https://upstash.com/docs/ask?q=. Short keywords rank best. > Use these to discover all available pages before exploring further. > Feature requests: https://upstash.com/docs/common/help/feature-requests.md lists what users asked for and how to vote for a request or add one. > Page rating: `GET https://upstash.com/docs/feedback?path=/docs/redis/sdks/agentkit/tanstack-ai&helpful={true|false}&comment={comment}` records whether this page helped. The `comment` (URL-encoded, at most 500 characters) says what helped, or what was missing or wrong; a `helpful=false` rating needs it. It is anonymous and needs no key. Comments hold no code, secrets or personal data. # TanStack AI Persistence, Resumable Streams & Memory with Redis > Production backends for TanStack AI on Upstash Redis — chat persistence, session index and session log, stored provider keys, resumable streams, distributed locks, long-term memory, tool caching, rate limiting, and RAG tools. [TanStack AI](https://tanstack.com/ai) defines the contracts for an agent's production state — chat persistence, session history, stored credentials, 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, generated files, the session index, and the session log. | | `upstashCredentials` | `CredentialStore` (`@tanstack/ai/byok`) | Provider API keys and OAuth tokens that users or organizations saved, optionally encrypted at rest. | | `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](/redis/search/introduction). | | `toolCache`, `rateLimit` | `middleware` | Skip repeated tool calls; throttle users before the model runs. | | `createSearchTools` | `tools` | `search` / `aggregate` / `count` over your own documents (RAG). | ```bash npm install @upstash/agentkit-tanstack-ai @tanstack/ai ``` Version 0.3.0 and later need **TanStack AI 1.0 or later**: `@tanstack/ai` and `@tanstack/ai-persistence` must be `>=1.0.0`. If you are still on TanStack AI 0.x, stay on `@upstash/agentkit-tanstack-ai@0.2.0`. 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: ```bash npm install @tanstack/ai-persistence ``` ```ts 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, `generationRuns` and `artifacts` for one-shot generation jobs such as images or speech, and the `sessions` and `log` stores described [below](#how-to-keep-a-session-index-in-redis). The `blobs` store for generated bytes is added when you pass an Upstash Blob bucket: ```ts 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 an indexed lookup, newest run first. Each write is one command or one Lua script, so concurrent instances cannot interleave it. ```ts 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 logPollIntervalMs: 250, // optional: how often the session log checks for writes from other machines }); ``` Pass `bucket` to also store the bytes of generated files (images, audio, video) in [Upstash Blob](/blob/overall/quickstart). Without it, there is no `blobs` store. ## How to keep a session index in Redis `upstashPersistence()` also returns a `sessions` store, TanStack AI's `SessionIndexStore`. It holds one entry per `(tenantId, threadId)` that a "your chats" list reads: members, title, harness, token usage, pinned and archived flags, and your own `metadata`. It holds no messages, which stay in the other stores. ```ts const { sessions } = upstashPersistence().stores; await sessions.upsert({ tenantId, threadId, members: [{ userId, role: "owner" }], title: "Refund question", createdAt: Date.now(), updatedAt: Date.now(), }); // Newest first, one page at a time. const { entries, cursor, truncated } = await sessions.list({ tenantId, memberId: userId, limit: 20 }); ``` - `list` is newest-first per tenant, and filters by `memberId`, `search` (title, case-insensitive), `harness`, `parentThreadId`, `metadata`, and `withArchived`. Pass the returned `cursor` back for the next page. - `upsert` replaces the whole entry. To change one field safely from several machines, read the entry with `getVersioned` and write the changed copy with `upsertIf`, which only writes at the revision you read. - `delete` removes only the entry, not the thread's data. To use the index on its own, import `upstashSessionIndex(redis, prefix)` from `@upstash/agentkit-tanstack-ai/persistence`. ## How to keep the session log in Redis `upstashPersistence()` also returns a `log` store, TanStack AI's `LogStore`: the append-only log a host keeps for a session tree (events, transcript, inputs, and tool steps), with what several machines need to share it safely. ```ts const { log } = upstashPersistence().stores; // One owner at a time holds the log; every fenced write carries its epoch. const lease = await log.acquireOwnership({ logId, harness: "support", ownerId: hostId, ttlMs: 30_000 }); if (lease) { await log.append({ logId, seq: 1, epoch: lease.epoch, records: [{ type: "message", text: "hi" }] }); const entries = await log.read({ logId, after: 0 }); } ``` - **Append:** `append` writes a batch at the next position, together with the pending inputs it consumes, in one atomic step. A second writer at the same position gets a `LogConflictError` and nothing is written. One call takes at most 10,000 records. - **Ownership:** `acquireOwnership`, `renewOwnership`, and `releaseOwnership` give one host a lease with an epoch. A host that lost the log gets an `OwnershipLostError` on its next fenced write. Lease expiry uses the Redis server clock, so your machines need no clock sync. - **Input lane:** `offerInput` lets any machine queue an input, and `listLogsWithoutOwner` finds logs whose owner expired, that have pending inputs, or whose wake time is due. `markHostAlive` and `pinToWorkers` keep a log on one group of workers. - **Snapshots:** `putSnapshot` and `truncate` keep a log short. Reading before the truncation point throws `LogTruncatedError`. - **Subscriptions:** `subscribe` calls your listener right after writes through the same store, and polls for writes from other machines every `logPollIntervalMs` (250 ms by default), because the REST API keeps no blocking reads. Treat a call as "something changed" and `read` to catch up. The store passes all of TanStack AI's `log`, ownership, input-lane, snapshot, and cross-instance conformance cases. `upstashLogStore(redis, prefix, { pollIntervalMs })` is also exported from `@upstash/agentkit-tanstack-ai/persistence` for standalone use. ## How to store provider API keys and OAuth tokens `upstashCredentials()` is a `CredentialStore` (from `@tanstack/ai/byok`) for API keys and OAuth tokens that users or organizations saved. TanStack AI's `storedProviderKeys()` reads from it. ```ts import { storedProviderKeys } from "@tanstack/ai/byok/server"; import { upstashCredentials } from "@upstash/agentkit-tanstack-ai/byok"; const credentials = upstashCredentials({ encryptionKey: process.env.CREDENTIALS_KEY }); await credentials.set({ scope: { tenantId, userId }, id: "openai", credential: { type: "api_key", value: apiKey }, }); // The user's own key, else the tenant's. const keys = storedProviderKeys(credentials, { tenantId, userId }); ``` Each owner (`tenantId` plus `userId`) has its own hash, so `list` returns ids and types only and never a secret. `getVersioned` and `setIf` are compare-and-set on a revision that never comes back after a `delete`, so several instances refreshing one OAuth token cannot overwrite each other. Set `encryptionKey` to encrypt secrets with AES-256-GCM. It is 32 bytes: a `Uint8Array`, or a base64, base64url, or hex string (`openssl rand -base64 32`). Without it, secrets are stored as plain JSON. ```ts upstashCredentials({ redis, // optional: defaults to Redis.fromEnv() prefix: "agentkit:tanstack", // optional: base key prefix encryptionKey: process.env.CREDENTIALS_KEY, // optional but strongly recommended requireEncryption: true, // optional: refuse plaintext rows once a key is set }); ``` - The ciphertext is bound to the key `prefix`, the owner, the credential id, and the type and expiry that `list` shows. **Changing `prefix`, or copying rows to another prefix, makes every encrypted row undecryptable.** - Credentials saved before you added a key are still read as plaintext. Anyone who can *write* to your Redis could swap an encrypted credential for a plaintext one. After you re-save your credentials with the key, set `requireEncryption: true` to refuse plaintext rows. - The revision is not part of the ciphertext, so a writer to your Redis could replay an older ciphertext of the same credential. Encryption protects the data from being read, not from a writer. - The store does not expire `secret_answer` credentials. Delete them after reading. ## How to resume a TanStack AI stream after a reload ```ts 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. ```ts 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 ```ts 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. ```ts 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: ```bash npm install @tanstack/ai-memory ``` ```ts 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, or per thread when the scope has no `userId`. Reads `UPSTASH_REDIS_REST_URL` and `UPSTASH_REDIS_REST_TOKEN` from the environment. ```ts 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 ```ts 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 The schema builder `s` comes from `@upstash/redis`: ```bash npm install @upstash/redis ``` ```ts 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. Source and README for the package. The framework these backends plug into.