DEV Community: Damla Hamurcu

I Let an LLM Play Cookie Clicker. Then I Fixed Its Biggest Weakness.

Damla Hamurcu — Wed, 01 Apr 2026 06:53:36 +0000

What a toy game taught me about when AI planning actually beats simple heuristics, and when it doesn't.

Cookie Clicker is secretly a resource allocation problem. You have one currency (cookies), a growing menu of buildings that generate cookies per second (CpS), tiered upgrades that double building output, and costs that scale exponentially. Every tick, you face the same decision: what should I buy next?

It feels like the kind of problem that rewards planning. Look ahead, sequence your purchases toward powerful upgrades, and you should come out ahead. So I built a simplified Cookie Clicker simulation in Python and asked: can an LLM make better spending decisions than a simple greedy algorithm?

The answer is more interesting than "yes" or "no." The LLM identifies opportunities the greedy algorithm is structurally blind to, but it executes them inside a worse control loop. Fix the control loop, and the combination beats both.

The Setup

The simulation has 10 buildings (Cursor through Alchemy Lab), tiered upgrades that unlock at ownership thresholds and double building output, grandma synergies that scale with building count, and standard Cookie Clicker cost scaling (each purchase multiplies the next by 1.15×). No golden cookies, no prestige, but the core resource allocation problem is intact.

I tested four strategies. All starting from the same mid-game checkpoint (tick 35,000, ~7M CpS) and running independently for 5,000 ticks.

Buy Cheapest always buys the cheapest affordable item. No evaluation, no prioritization. The baseline.

Greedy ROI evaluates every available purchase by its payback period: how many ticks until this purchase pays for itself in additional CpS? It picks the shortest payback time, every tick. No memory. No planning. Just the locally optimal choice, relentlessly.

LLM Planner receives the full game state: current CpS, every building count, every available upgrade with its cost and CpS impact, and nearby tier-unlock thresholds. It produces a prioritized 10-purchase plan with reasoning. It re-plans when its queue empties or when new upgrades unlock. I used GPT-5-mini with structured output.

Hybrid uses the LLM planner for strategic direction, but falls back to the greedy ROI algorithm when the LLM is waiting to afford its next planned purchase. About 30 lines of wrapper code.

Results

Strategy	Final CpS	Purchases	Upgrades	LLM Calls
Hybrid	8,911,796	30	1	2
Greedy ROI	8,771,218	27	0	0
LLM Planner	8,314,442	42	1	5
Buy Cheapest	7,573,220	177	2	0

The hybrid wins. It beats pure greedy by 1.6% and the pure LLM by 7.2%.

1.6% sounds small, and a skeptical reader might think "that's noise." But it was consistent across runs, and it came from only 2-4 LLM calls total. The value isn't coming from continuous planning. It's coming from a small number of well-timed strategic interventions.

The final numbers don't tell the real story though. The CpS-over-time chart does.

What Actually Happened

Three distinct behaviors show up in the data.

Greedy ROI makes a purchase roughly every 150-200 ticks. Each one is locally optimal. It never pauses, never wastes cookies, and never makes a bad buy. Its CpS line is a steady staircase upward.

LLM Planner starts strong. It identified that pushing Factory count from 45 to 50 would unlock a tier 4 upgrade that doubles all Factory output. It queued five consecutive Factory purchases followed by the upgrade. Here's its reasoning on Factory #48:

"Still driving toward the Factory tier 4 unlock; each Factory bought is invested toward a guaranteed doubling once the tier is purchased."

This is a genuinely smart move. The greedy algorithm would never buy five Factories in a row when a single Temple has a shorter payback period, even though the five Factories plus their unlock are worth far more in aggregate.

But after executing its plan, the LLM went silent. Its queue emptied and the re-planning trigger didn't fire. Cookies accumulated with nothing to spend them on. CpS flatlined from tick 3,300 to tick 5,000. That's nearly a third of the entire run doing nothing. By the time greedy caught up and passed it, the damage was done.

The Hybrid got the best of both. It executed the same Factory threshold push (ticks 1,258-1,398 in its run), then let greedy take over during the gaps. Look at ticks 1,431-1,675: a burst of Grandma, Temple, Cursor, Farm, and Mine purchases. That's the greedy fallback spending cookies productively while the LLM queue was empty. No stalling. No idle cookies.

Why Greedy Is Secretly Strong

Here's the part that surprised me, and the real insight from this experiment.

The greedy algorithm isn't naive. When it evaluates a purchase, it calls _cps_delta(), which simulates the full game state after the purchase. That includes tier upgrade doublings, grandma synergy multipliers, everything. The moment a tier upgrade becomes available, greedy sees its full multiplicative effect as an immediate CpS delta and buys it if the payback is good.

The environment is doing something subtle here: it collapses long-term value into short-term signal. Greedy doesn't need to "plan" for a tier upgrade because the upgrade's value is fully visible the tick it unlocks. Greedy is really "greedy over a rich instantaneous reward function," and that's a much harder baseline to beat than it looks. This also favors greedy more than many real-world systems, where the impact of decisions is often delayed or only partially observable.

The only place greedy is structurally blind is threshold plays: sequences of purchases that are individually suboptimal but unlock something powerful at the end. Buying five Factories when each one has worse ROI than a Temple requires looking ahead. Greedy can't do that. The LLM can.

The LLM Doesn't Even Agree With Itself

I ran the hybrid twice from the same starting state. Run 1: the LLM pushed Factories to 50 for the tier 4 unlock (2 calls, 8.91M CpS). Run 2: it targeted Alchemy Lab and Shipment scaling (4 calls, roughly the same final CpS).

Completely different plans. Same outcome.

This tells you something important: the hybrid architecture is robust to LLM variance. The greedy fallback acts as a stabilizer. Any halfway-decent strategic direction gets converted into consistent execution. The system doesn't require a consistently good plan. Just a non-random one.

This is the opposite of the pure LLM, where different plans produced meaningfully different results (8.3M vs 7.9M in earlier testing). Without the greedy safety net, plan quality matters a lot. With it, plan quality matters much less.

The Control Loop Matters More Than the Intelligence

The pure LLM's problem was never intelligence. It identified the Factory threshold push, sequenced synergy upgrades correctly, and reasoned about multi-step interactions in a complex system. Its problem was architectural. It was a blocking queue executor with no interrupt mechanism.

# Pure LLM: blocks when queue is empty
if not self._plan:
    self._plan = self._call_llm(state)  # expensive, slow
# Nothing happens between plans

# Hybrid: fills gaps with greedy
action = self._llm.decide(state)
if action is not None:
    return action
return self._greedy_decide(state)  # never idle

The greedy algorithm re-evaluates every tick. The LLM hard-blocks until its planned item is affordable. That's not a fair comparison of planning vs. heuristics. It's a comparison of a continuous control loop vs. a batch executor.

Fixing this required about 30 lines of code. The hybrid wrapper calls the LLM planner first. If it returns nothing (waiting to afford something), it runs greedy ROI as a fallback, with one constraint: don't spend cookies that the LLM's next planned purchase needs. This prevents the fallback from interfering with planned threshold plays, but it also introduces small inefficiencies when the planned purchase is far away. More on that below.

What This Means Beyond Cookie Clicker

I see the same dynamic in production systems constantly.

Simple, reliable heuristics often outperform AI-driven approaches. Not because they're smarter, but because they're predictable and they never stop working. A hand-tuned rule that runs on every request beats a sophisticated model that occasionally hangs, returns malformed output, or needs a retry loop.

But the answer isn't "don't use AI." The answer is: use AI for what it's good at and heuristics for what they're good at. The LLM is good at spotting structural opportunities across a complex state space. The greedy algorithm is good at relentless, consistent execution. Neither is good at the other's job.

The hybrid pattern generalizes:

Let the AI set strategic direction periodically (identify thresholds, sequence multiplier chains, spot non-obvious opportunities)
Let a simple, fast heuristic handle moment-to-moment execution
Never let the system sit idle because the AI hasn't spoken

This is how I'd architect any system where you're tempted to put an LLM in a hot loop: don't. Put it in a planning layer that fires occasionally, and let deterministic logic handle the rest.

What I'd Do Next

The hybrid's reservation logic is still too conservative. It hoards cookies for planned purchases even when they're thousands of ticks away, creating mini-stalls. A smarter version would only reserve cookies when the planned purchase is close to affordable.

More interesting: what happens if you break greedy's advantage? The reason greedy is so strong here is that _cps_delta() makes long-term value immediately visible. What if upgrades had hidden effects? What if some rewards were delayed? That's where I'd expect the LLM's planning advantage to widen significantly. And it's the experiment I'm most excited to try next.

Try It Yourself

The full simulation, all four strategies, and the comparison charts are on GitHub: https://github.com/hamurda/cookie-clicker-strategy-lab.

If you've benchmarked LLMs against algorithmic baselines on other problems, I'd love to hear what you found. Does the "strategic planning + consistent execution" pattern show up in your work too?

What Cookie Clicker Taught Me About AI-Assisted Development

Damla Hamurcu — Mon, 23 Mar 2026 00:58:54 +0000

I was playing Cookie Clicker, the idle game where you buy buildings that generate cookies, which you spend on more buildings, and I kept hitting genuine dilemmas. Do I buy the cheap cursor now, or save for the grandma that compounds better over time? Is there a mathematically best approach, or is this just vibes?

That's when I realized: this is a resource allocation problem with compounding returns and opportunity costs. Could an LLM reason about it? And could I build the whole thing using Claude Code?

Four sessions over a week, using Claude Code with Sonnet 4.6, I built a Cookie Clicker simulation engine that pits a greedy algorithmic strategy against an LLM-powered planner. Here's what I learned.

The Project

I'm an engineer building AI projects on the side. I'm not a senior software engineer, and that matters for this story, because a lot of what I learned came from the gaps between what Claude Code produced and what I could evaluate.

The simulation engine runs a simplified Cookie Clicker: buildings generate cookies per second (CpS), costs scale with each purchase, and different strategies compete over a fixed number of ticks. One strategy uses a greedy ROI formula. The other asks an LLM to plan its purchases. Before opening Claude Code, I wrote a detailed spec covering game mechanics, cost formulas, upgrade synergies, and what the output should look like. This turned out to be the most important decision of the project.

Session 1 & 2: Spec-First Works — Until You Refactor

In the first session, I gave Claude Code the spec and asked it to build the engine. It produced working code quickly. The core math (cost scaling, CpS calculation, tier multipliers, grandma synergy stacking) was correct on the first pass. The file structure was clean: data at the top, pure functions in the middle, simulation loop at the bottom.

The pattern: spec quality determines output quality. When I gave Claude Code a detailed, unambiguous spec, it delivered. When I was vague, I got code that worked but needed rethinking.

But even with a good spec, I noticed a habit. When I asked for an end-of-simulation report, Claude Code wrote fresh functions to gather the data it needed, even though I'd already built functions that computed the same values for the simulation loop. It added a total_baked field to the game state dataclass and never referenced it anywhere. Maybe Claude Code was testing me too. Who knows. It solves each new requirement independently rather than looking at what already exists.

Session 2 is where things got interesting. I asked Claude Code to refactor the single file into three: game engine, strategies, and simulation runner. It produced a clean architecture. A Strategy abstract base class with a single method, typed return values, proper separation. The dependency graph had no cycles. Textbook stuff.

Then I ran the code and found bugs I'd already fixed. The same ones. Manually. In the previous session.

In session 1, I'd manually corrected a double-computation issue in the simulation loop and fixed a reporting function that was recomputing values instead of receiving them. During the refactor, Claude Code put both problems back. It doesn't have memory of why your code looks the way it does. Every refactor is a fresh start from its understanding of the goal, not from your history of deliberate improvements.

And so I went looking for solutions. Could I write better CLAUDE.md directions to preserve certain patterns? More explicit prompting? I tried restricting what it was allowed to change, but that created a different problem: Claude Code became a typing machine, just executing instructions without the design intelligence that makes it useful in the first place. There's a tension between giving it freedom to make good decisions and preventing it from undoing yours.

What actually helped was simpler: start a fresh session for each phase, with a clear recap of what's done and what's next. A fresh context with a focused brief ("here's the codebase, here's what's working, here's the one thing I need you to do") consistently produced better results than continuing a session full of corrections and pivots.

This became my most important rule: after any structural refactor, the first thing you check is whether it undid something you already fixed.

Session 3: Where Spec-First Hit Its Limits

Building the LLM strategy required a different kind of work. I couldn't just write a spec and hand it off. The design problem was genuinely iterative.

The architecture I landed on was a planner: ask the LLM to plan the next 10 purchases given the full game state, execute them sequentially, and re-plan when the list runs out. Claude Code implemented this cleanly. But the behavior surprised me.

The LLM had everything it needed. Ticks remaining, current CpS, how each purchase would affect CpS, all available options and their costs. And yet, it planned to waste the last 1,700 ticks doing nothing. It identified an expensive building as the best ROI, committed to saving for it, and didn't buy anything else while waiting. When I increased the call frequency, asking for only the next purchase, it performed noticeably better. Higher API bills, better results. Apparently money talks.(And that's another post's discussion.)

But here's why this matters for the Claude Code story. Even with perfect information in a fully deterministic system, the LLM needed regular check-ins to stay on track. Now imagine giving Claude Code all your project requirements upfront and expecting it to execute a multi-phase architecture without course correction. If deterministic math problems need re-planning, software design certainly does. This was the moment where I couldn't just write a better spec. The spec was fine. The implementation was correct. The problem was a design tradeoff I could only discover by running the code and being surprised.

Using the Tool to Check the Tool

Session 3 also surfaced a trust problem. Claude Code was producing dead code, unused variables, unreachable if-blocks, even when following my directions precisely. I caught some in review. I caught others by mentally running the logic. But I'm not senior enough to be confident I was catching everything, and I don't have a senior engineer to lean on.

So I tried something: in a fresh session, I asked Claude Code to review the codebase like an experienced mentor and write its findings to a review.md file. It produced a structured list of issues, categorized by severity. Most findings were cosmetic: missing docstrings, suggestions to rename variables. But buried in there was a real one. A dead if-block in the game logic, an unreachable branch that I'd walked right past during my own review.

I curated the list. Removed the noise, kept the findings I agreed with. Then I opened another fresh session and told it to fix the issues listed in review.md. It did, cleanly. No side effects, no "improvements" I didn't ask for.

The workflow: build → new session to review → curate findings → new session to fix. Each session has a single, well-defined job. The context stays clean. You can extend this to architecture review, test generation, or documentation. The key insight is that review and implementation should never happen in the same session. Mixing them is how you get silent regressions.

The Patterns

After four sessions, here's my working model for Claude Code:

Phase the work. Small, clear, well-defined bites. One session per phase, with a fresh context and a clear brief. This solves the majority of issues I encountered: lost intent, dead code accumulation, silent regressions. There's evidence for this in the project itself. In session 4, I gave Claude Code a precise architectural refactor: exact folder structure, where each file should go, what should change and what shouldn't. It executed perfectly. No dead code, no regressions, no unsolicited improvements. The same tool that generated unused dataclass fields during open-ended feature work produced spotless output when the task was bounded and clear.

After any refactor, check for regressions. This is the number one failure mode. Claude Code doesn't carry your design intent across structural changes. It rebuilds from its understanding, not your history. Treat every refactor output like a code review, not a delivery.

Give the spec, not the steps. Tell Claude Code what should happen, not how to implement it. It sometimes chose simpler representations than what I'd planned, and they were better. Leave room for it to make implementation decisions.

You decide architecture. If you're delegating decisions you don't understand, stop and think. Claude Code implements faithfully, but it won't push back and say "this structure won't scale" or "you'll regret this when you add feature X." That's your job.

The less you know, the more it costs. When I knew exactly what the output should look like, Claude Code was fast and I could verify the results in minutes. When I was in unfamiliar territory, I burned through iterations, tokens, and time. The cost of using these tools scales with your uncertainty. The clearer your mental model of what "done" looks like, the faster and cheaper you get there. Claude Code doesn't close that gap. It makes the gap cheaper to navigate, but the gap is still yours.

The Judgment Layer

Claude Code is a strong first-draft tool. It gets mechanics right, produces clean structure, and follows specs faithfully. But it doesn't carry forward your design intent across refactors. It doesn't anticipate how code needs to evolve. And it won't push back on decisions that will cause problems later.

What you bring as the engineer is the judgment layer: reviewing for regressions, catching duplicated logic, making structural decisions about what the code needs to become, and iterating on design tradeoffs that require understanding the why behind the code, not just the what.

For a small project like this, I could see the right architecture. The question I'm still sitting with: what happens when I can't?

The full project is on GitHub. Built with Claude Code (Sonnet 4.6), four sessions, roughly a week of part-time work.

Build an MCP Server That Finds Your RAG Chatbot's Blind Spots

Damla Hamurcu — Mon, 16 Mar 2026 01:36:23 +0000

RAG chatbots fail silently.

Imagine you've hired a chatbot to handle FAQs on your behalf, so you can level up your D&D character instead of replying to the same questions over and over. You provided all the information you could think of.

Then your potential biggest client asks if you're left or right handed. The chatbot doesn't know. The client walks away. It never occurred to you to teach it. Who asks that? You will never know. Because you had no idea there was a missed question in the first place.

In this tutorial, we're building an MCP server that creates that feedback loop. When the chatbot gives a low-confidence answer, it logs the question. Over time, questions pile up. Run the pattern analysis and it clusters them by semantic similarity — "oh, 40 people asked about refund policies in different ways, and we have zero documentation on that." Then it suggests what to write, and you close the loop.

We'll build it step by step. By the end, you'll have a working MCP server with four tools:

Tool	What it does
`log_unanswered_question`	Store a question + its embedding
`get_question_patterns`	Cluster unresolved questions, return topics
`suggest_documents`	AI-generated doc outline for a topic
`mark_resolved`	Close the loop after adding documentation

Let's start with the simplest possible version and layer on complexity as we go.

Repo: github.com/hamurda/rag-insights-mcp

Setup

mkdir unanswered-questions-mcp && cd unanswered-questions-mcp
uv init && uv add mcp[cli] openai numpy scikit-learn aiosqlite pydantic python-dotenv

Create a .env file:

OPENAI_API_KEY=sk-your-key-here

And a config.py to load it:

import os
from pathlib import Path
from dotenv import load_dotenv

load_dotenv()


class Config:
    DATABASE_PATH: str = os.getenv("DATABASE_PATH", "unanswered_questions.db")
    OPENAI_API_KEY: str = os.getenv("OPENAI_API_KEY", "")
    EMBEDDING_MODEL: str = os.getenv("EMBEDDING_MODEL", "text-embedding-3-small")
    CHAT_MODEL: str = os.getenv("CHAT_MODEL", "gpt-5-mini")
    SIMILARITY_THRESHOLD: float = float(os.getenv("SIMILARITY_THRESHOLD", "0.50"))
    MIN_CLUSTER_SIZE: int = int(os.getenv("MIN_CLUSTER_SIZE", "3"))
    MAX_CLUSTER_SIZE: int = int(os.getenv("MAX_CLUSTER_SIZE", "20"))
    EMBEDDING_BATCH_SIZE: int = int(os.getenv("EMBEDDING_BATCH_SIZE", "100"))

    @classmethod
    def validate(cls) -> None:
        if not cls.OPENAI_API_KEY:
            raise ValueError("OPENAI_API_KEY is required. Set it in .env or environment.")
        if not 0 <= cls.SIMILARITY_THRESHOLD <= 1:
            raise ValueError(f"SIMILARITY_THRESHOLD must be 0-1, got {cls.SIMILARITY_THRESHOLD}")

    @classmethod
    def db_path(cls) -> Path:
        return Path(__file__).parent / cls.DATABASE_PATH

Everything is configurable via environment variables, but the defaults are fine for now. That SIMILARITY_THRESHOLD of 0.50 looks low, and there's a story behind it.

Step 1: Store a Question

First things first. We need a place to put questions. Let's start with the database.

Because we want to save the questions and group the common ones, we need two tables: questions for individual logged questions, and clusters for the patterns we'll discover later.

And because this MCP Server is built for a multi-tenant chatbot, every table needs a tenant_id column. If you're running a single chatbot, leave it None, but if you're building for multiple customers, tenant scoping is baked in from the start. Much easier than retrofitting.

# database.py
import json
import uuid
import pickle
from datetime import datetime, timezone
from typing import Optional

import aiosqlite

from config import Config

SCHEMA = """
CREATE TABLE IF NOT EXISTS questions (
    id TEXT PRIMARY KEY,
    question TEXT NOT NULL,
    context TEXT,
    tenant_id TEXT,
    metadata TEXT,
    embedding BLOB,
    created_at TEXT NOT NULL,
    resolved INTEGER DEFAULT 0,
    resolved_at TEXT,
    resolved_by TEXT,
    resolution_notes TEXT
);

CREATE TABLE IF NOT EXISTS clusters (
    id TEXT PRIMARY KEY,
    tenant_id TEXT,
    topic TEXT,
    question_ids TEXT,
    representative_questions TEXT,
    created_at TEXT NOT NULL
);

CREATE INDEX IF NOT EXISTS idx_questions_tenant ON questions(tenant_id);
CREATE INDEX IF NOT EXISTS idx_questions_resolved ON questions(resolved);
CREATE INDEX IF NOT EXISTS idx_questions_created ON questions(created_at);
"""

Why SQLite? Because this isn't real-time retrieval. Nobody is searching the vector space at query time — we're doing batch analysis on a schedule. SQLite is zero configuration, single file, and handles 10K+ questions without breaking a sweat.

Now the Database class. We'll start with just enough to save and retrieve questions:

class Database:
    def __init__(self, db_path: Optional[str] = None):
        self.db_path = db_path or str(Config.db_path())
        self._conn: Optional[aiosqlite.Connection] = None

    async def connect(self) -> None:
        self._conn = await aiosqlite.connect(self.db_path)
        await self._conn.executescript(SCHEMA)

    async def close(self) -> None:
        if self._conn:
            await self._conn.close()

    async def save_question(
        self, question: str, context: str | None = None,
        tenant_id: str | None = None, metadata: dict | None = None,
        embedding: list[float] | None = None,
    ) -> str:
        qid = str(uuid.uuid4())
        await self._conn.execute(
            """INSERT INTO questions
               (id, question, context, tenant_id, metadata, embedding, created_at)
               VALUES (?, ?, ?, ?, ?, ?, ?)""",
            (
                qid, question, context, tenant_id,
                json.dumps(metadata) if metadata else None,
                pickle.dumps(embedding) if embedding else None,
                datetime.now(timezone.utc).isoformat(),
            ),
        )
        await self._conn.commit()
        return qid

    async def get_unresolved(
        self, tenant_id: str | None = None, limit: int | None = None
    ) -> list[dict]:
        query = "SELECT * FROM questions WHERE resolved = 0"
        params: list = []
        if tenant_id:
            query += " AND tenant_id = ?"
            params.append(tenant_id)
        query += " ORDER BY created_at DESC"
        if limit:
            query += " LIMIT ?"
            params.append(limit)
        cursor = await self._conn.execute(query, params)
        return [self._to_dict(row) for row in await cursor.fetchall()]

    async def get_question(self, question_id: str) -> dict | None:
        cursor = await self._conn.execute(
            "SELECT * FROM questions WHERE id = ?", (question_id,)
        )
        row = await cursor.fetchone()
        return self._to_dict(row) if row else None

    @staticmethod
    def _to_dict(row) -> dict:
        return {
            "id": row[0], "question": row[1], "context": row[2],
            "tenant_id": row[3],
            "metadata": json.loads(row[4]) if row[4] else None,
            "embedding": pickle.loads(row[5]) if row[5] else None,
            "created_at": row[6], "resolved": bool(row[7]),
            "resolved_at": row[8], "resolved_by": row[9],
            "resolution_notes": row[10],
        }

Embeddings are stored as pickled BLOBs. That feels wrong, I know. We never query by embedding. There's no WHERE embedding NEAR ... happening here. We load all unresolved questions into memory and compute pairwise similarities. Pickle is the simplest way to get a list[float] in and out of SQLite, and for this access pattern, it earns its place.

Let's verify it works. Create a quick test script:

# test_step1.py
import asyncio
from database import Database

async def main():
    db = Database("test.db")
    await db.connect()

    qid = await db.save_question(
        question="How do I reset my password?",
        context="Login page",
        tenant_id="acme",
    )
    print(f"Saved: {qid}")

    questions = await db.get_unresolved(tenant_id="acme")
    print(f"Retrieved: {questions[0]['question']}")

    await db.close()

asyncio.run(main())

$ python test_step1.py
Saved: a1b2c3d4-...
Retrieved: How do I reset my password?

Questions go in, questions come out. Now let's make them searchable.

Step 2: Add Embeddings

A question stored as text is just a string. To us, "how do I reset my password" and "I forgot my password, how do I change it" are obviously the same topic. To a database, they're two completely different rows. We need to turn them into vectors to find the pattern.

Create analyzer.py. We'll start with just the embedding logic:

# analyzer.py
import openai
from config import Config
from database import Database


class QuestionAnalyzer:
    def __init__(self, database: Database):
        self.db = database
        self.client = openai.AsyncOpenAI(api_key=Config.OPENAI_API_KEY)

    async def generate_embedding(self, text: str) -> list[float]:
        response = await self.client.embeddings.create(
            model=Config.EMBEDDING_MODEL, input=text
        )
        return response.data[0].embedding

Every question gets embedded when it's logged. But what about questions that were logged without embeddings? Maybe the API was down, maybe they came from a bulk import.

For that, we add a batch embedding method and a safety net that checks for gaps before analysis. The batching matters; OpenAI's embedding API accepts multiple inputs per call, so we chunk at 100 instead of making one API call per question.

    async def generate_embeddings_batch(self, texts: list[str]) -> list[list[float]]:
        if not texts:
            return []
        embeddings: list[list[float]] = []
        for i in range(0, len(texts), Config.EMBEDDING_BATCH_SIZE):
            batch = texts[i : i + Config.EMBEDDING_BATCH_SIZE]
            response = await self.client.embeddings.create(
                model=Config.EMBEDDING_MODEL, input=batch
            )
            embeddings.extend(item.embedding for item in response.data)
        return embeddings

    async def ensure_embeddings(self, questions: list[dict]) -> list[dict]:
        needs = [q for q in questions if q["embedding"] is None]
        if not needs:
            return questions
        texts = [q["question"] for q in needs]
        embeddings = await self.generate_embeddings_batch(texts)
        for q, emb in zip(needs, embeddings):
            q["embedding"] = emb
            await self.db.update_embedding(q["id"], emb)
        return questions

This checks every question before analysis and fills in anything that's missing. We'll also need update_embedding on the database, add this to database.py:

    async def update_embedding(self, question_id: str, embedding: list[float]) -> bool:
        cursor = await self._conn.execute(
            "UPDATE questions SET embedding = ? WHERE id = ?",
            (pickle.dumps(embedding), question_id),
        )
        await self._conn.commit()
        return cursor.rowcount > 0

Now when we log a question, we can generate the embedding at the same time:

# test_step2.py
import asyncio
from database import Database
from analyzer import QuestionAnalyzer

async def main():
    db = Database("test.db")
    await db.connect()
    analyzer = QuestionAnalyzer(db)

    questions = [
        "How do I reset my password?",
        "I forgot my password, how do I change it?",
        "Where can I find my invoice?",
    ]

    for q in questions:
        embedding = await analyzer.generate_embedding(q)
        qid = await db.save_question(question=q, embedding=embedding, tenant_id="acme")
        print(f"Saved '{q[:40]}' with {len(embedding)}-dim embedding")

    await db.close()

asyncio.run(main())

$ python test_step2.py
Saved 'How do I reset my password?' with 1536-dim embedding
Saved 'I forgot my password, how do I change it' with 1536-dim embedding
Saved 'Where can I find my invoice?' with 1536-dim embedding

Each question is now a 1536-dimensional vector. Time to compare them.

Step 3: Cluster by Similarity

This is the core of the analysis. We have questions as vectors — now we need to find which ones are about the same topic.

The approach: compute pairwise cosine similarities, then greedily assign questions to clusters. Add this to QuestionAnalyzer:

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

    def cluster_questions(
        self, questions: list[dict], threshold: float | None = None
    ) -> list[list[dict]]:
        if not questions:
            return []

        threshold = threshold or Config.SIMILARITY_THRESHOLD
        embeddings = np.array([q["embedding"] for q in questions])
        sim = cosine_similarity(embeddings)

        clusters: list[list[dict]] = []
        assigned: set[int] = set()

        for i, q in enumerate(questions):
            if i in assigned:
                continue
            cluster = [q]
            assigned.add(i)
            for j in range(len(questions)):
                if j not in assigned and sim[i][j] >= threshold:
                    cluster.append(questions[j])
                    assigned.add(j)
            if len(cluster) >= Config.MIN_CLUSTER_SIZE:
                clusters.append(cluster[: Config.MAX_CLUSTER_SIZE])

        return clusters

Walk through questions in order. For each unassigned question, start a new cluster. Then scan every other unassigned question. If its cosine similarity to the first one exceeds the threshold, pull it in. Once you've scanned everything, move to the next unassigned question and start a new cluster. If that doesn't click on the first read, tell it to your rubber duck.

Let's test it with real data. Here are 14 questions: three obvious clusters and two singletons:

# test_step3.py
import asyncio
from database import Database
from analyzer import QuestionAnalyzer

SAMPLE_QUESTIONS = [
    # Password reset cluster
    {"question": "How do I reset my password?", "tenant_id": "acme"},
    {"question": "I need to reset my password", "tenant_id": "acme"},
    {"question": "I forgot my password, how do I change it?", "tenant_id": "acme"},
    {"question": "My password isn't working, how do I reset it?", "tenant_id": "acme"},
    # Billing cluster
    {"question": "Where can I find my invoice?", "tenant_id": "acme"},
    {"question": "How do I download my invoice?", "tenant_id": "acme"},
    {"question": "I need a copy of my invoice", "tenant_id": "acme"},
    {"question": "Can I get a receipt for my payment?", "tenant_id": "acme"},
    # API rate limits cluster
    {"question": "What are the API rate limits?", "tenant_id": "techcorp"},
    {"question": "How many API requests can I make per minute?", "tenant_id": "techcorp"},
    {"question": "I'm hitting the API rate limit, what's the cap?", "tenant_id": "techcorp"},
    {"question": "Is there a limit on API calls?", "tenant_id": "techcorp"},
    # Singletons — should NOT cluster
    {"question": "Do you support single sign-on?", "tenant_id": "techcorp"},
    {"question": "What colors can I use for my dashboard theme?", "tenant_id": "acme"},
]

async def main():
    db = Database("test_clustering.db")
    await db.connect()
    analyzer = QuestionAnalyzer(db)

    # Log all questions with embeddings
    for q in SAMPLE_QUESTIONS:
        emb = await analyzer.generate_embedding(q["question"])
        await db.save_question(question=q["question"], tenant_id=q["tenant_id"], embedding=emb)

    # Retrieve and cluster
    questions = await db.get_unresolved()
    clusters = analyzer.cluster_questions(questions)

    print(f"Found {len(clusters)} cluster(s):\n")
    for i, cluster in enumerate(clusters):
        print(f"Cluster {i + 1} ({len(cluster)} questions):")
        for q in cluster:
            print(f"  - {q['question']}")
        print()

    await db.close()

asyncio.run(main())

Run it and you should see three clusters. The password questions find each other. The billing questions find each other. The API rate limit questions find each other. The SSO and dashboard theme questions sit alone and get dropped because they don't meet MIN_CLUSTER_SIZE of 3.

About that threshold

That SIMILARITY_THRESHOLD of 0.50 looks low. I started at 0.70, thinking that's a reasonable bar for "these questions are about the same thing."

It wasn't. The problem is the greedy algorithm: the first question in each cluster becomes the anchor, and every other question is measured against it. Not against the cluster centroid, not against the closest member. So if the anchor is "How do I reset my password?" and another question is "What's your policy on account recovery?", text-embedding-3-small might score that at 0.55. Same topic to a human. Below threshold at 0.70.

Dropping to 0.50 fixed it. The safety net is MIN_CLUSTER_SIZE. A loose threshold won't generate noise because stray pairs get dropped. You need at least 3 questions about the same thing before it's reported as a pattern.

Could I have used HDBSCAN instead? It doesn't depend on anchor order and it's more robust. But it felt like bringing a machine learning framework to a problem I could solve with 30 lines of code and a lower threshold. If the clusters start looking noisy with real production data, HDBSCAN is the upgrade path.

Step 4: Complete the analysis pipeline

We can cluster. Now we need to turn clusters into something useful.

First, a way to pick the most representative questions from a cluster, the ones most similar to everything else:

# Add to QuestionAnalyzer in analyzer.py

    def pick_representative(self, cluster: list[dict], n: int = 3) -> list[str]:
        if len(cluster) <= n:
            return [q["question"] for q in cluster]

        embeddings = np.array([q["embedding"] for q in cluster])
        sim = cosine_similarity(embeddings)
        centrality = [
            (np.mean([sim[i][j] for j in range(len(cluster)) if j != i]), cluster[i])
            for i in range(len(cluster))
        ]
        centrality.sort(key=lambda x: x[0], reverse=True)
        return [q["question"] for _, q in centrality[:n]]

Next, topic labeling. This is where things get architecturally interesting. The MCP tool itself makes an LLM call:

    async def extract_topic(self, cluster: list[dict]) -> str:
        sample = [q["question"] for q in cluster[:10]]
        prompt = (
            "These are similar customer questions. "
            "Reply with ONLY a 2-5 word topic label.\n\n"
            + "\n".join(f"- {q}" for q in sample)
        )
        resp = await self.client.chat.completions.create(
            model=Config.CHAT_MODEL,
            messages=[{"role": "user", "content": prompt}],
        )
        return resp.choices[0].message.content.strip()

An LLM tool that calls another LLM. I have thoughts on this but that's a separate post. For now, it works, and the alternative (hand-labeling clusters) doesn't scale.

Now the orchestration method that ties it all together — fetch unresolved questions, ensure embeddings, cluster, and label:

    async def find_patterns(
        self, tenant_id: str | None = None, min_cluster_size: int | None = None,
    ) -> list[dict]:
        questions = await self.db.get_unresolved(tenant_id=tenant_id)
        if not questions:
            return []

        questions = await self.ensure_embeddings(questions)
        clusters = self.cluster_questions(questions)

        if min_cluster_size:
            clusters = [c for c in clusters if len(c) >= min_cluster_size]

        patterns: list[dict] = []
        for cluster in clusters:
            topic = await self.extract_topic(cluster)
            representative = self.pick_representative(cluster)
            qids = [q["id"] for q in cluster]

            patterns.append({
                "topic": topic,
                "count": len(cluster),
                "representative_questions": representative,
                "question_ids": qids,
                "earliest_date": min(q["created_at"] for q in cluster),
                "latest_date": max(q["created_at"] for q in cluster),
            })
            await self.db.save_cluster(
                topic=topic, question_ids=qids,
                representative_questions=representative, tenant_id=tenant_id,
            )

        patterns.sort(key=lambda p: p["count"], reverse=True)
        return patterns

We also need save_cluster, mark_resolved and get_stats on the database, plus suggest_documentation on the analyzer that powers the suggest_documents tool. These are straightforward: full source on GitHub. Here's mark_resolved since it completes the feedback loop:

# Add to Database in database.py

    async def mark_resolved(
        self, question_id: str, resolved_by: str, notes: str | None = None
    ) -> bool:
        cursor = await self._conn.execute(
            """UPDATE questions
               SET resolved = 1, resolved_at = ?, resolved_by = ?, resolution_notes = ?
               WHERE id = ?""",
            (datetime.now(timezone.utc).isoformat(), resolved_by, notes, question_id),
        )
        await self._conn.commit()
        return cursor.rowcount > 0

Step 5: Wrap It in MCP

We have a working system: questions go in with embeddings, clustering finds patterns. Now let's expose it as an MCP server so any MCP client (Claude Desktop, an agent, a scheduled script) can use it.

Now server.py. FastMCP gives us a clean way to define tools with typed inputs:

# server.py
import json
from contextlib import asynccontextmanager
from typing import Optional

from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field

from config import Config
from database import Database
from analyzer import QuestionAnalyzer


@asynccontextmanager
async def lifespan(server: FastMCP):
    Config.validate()
    db = Database()
    await db.connect()
    analyzer = QuestionAnalyzer(db)
    yield {"db": db, "analyzer": analyzer}
    await db.close()


mcp = FastMCP("unanswered_questions_mcp", lifespan=lifespan)

The lifespan context manager creates one database connection and one analyzer for the server's lifetime. Every tool handler gets them from ctx.request_context.lifespan_state. No global singletons, clean shutdown.

Each tool gets a Pydantic input model. These descriptions aren't just for docs. They're what the MCP client sees when deciding how to call the tool:

class LogQuestionInput(BaseModel):
    question: str = Field(..., description="The question the chatbot couldn't answer")
    context: Optional[str] = Field(None, description="Where/why this question was asked")
    tenant_id: Optional[str] = Field(None, description="Tenant identifier for multi-tenant systems")
    metadata: Optional[dict] = Field(None, description="Extra metadata (user_id, confidence, source, etc.)")

And the tool itself:

@mcp.tool(
    name="log_unanswered_question",
    annotations={
        "title": "Log Unanswered Question",
        "readOnlyHint": False,
        "destructiveHint": False,
        "idempotentHint": False,
        "openWorldHint": True,
    },
)
async def log_unanswered_question(params: LogQuestionInput, ctx=None) -> str:
    db: Database = ctx.request_context.lifespan_state["db"]
    analyzer: QuestionAnalyzer = ctx.request_context.lifespan_state["analyzer"]

    embedding = await analyzer.generate_embedding(params.question)
    qid = await db.save_question(
        question=params.question,
        context=params.context,
        tenant_id=params.tenant_id,
        metadata=params.metadata,
        embedding=embedding,
    )
    return json.dumps({"success": True, "question_id": qid})

The annotations tell the MCP client what kind of operation this is. readOnlyHint: False because it writes data. idempotentHint: False, since calling it twice creates two records. openWorldHint: True, as it calls external services.

The other three tools follow the same pattern: get_question_patterns calls find_patterns and includes stats, suggest_documents generates doc outlines, and mark_resolved closes the loop. Full server source on GitHub.

Start the server:

if __name__ == "__main__":
    mcp.run()

Add it to Claude Desktop:

{
  "mcpServers": {
    "unanswered-questions": {
      "command": "python",
      "args": ["/path/to/unanswered-questions-mcp/server.py"]
    }
  }
}

Step 6: The Full Loop

Let's run through the entire lifecycle. The test suite (test_server.py) does this without starting the MCP server — same code paths, no transport layer in the way:

python test_server.py

Here's what happens:

1. Log 14 questions — 4 about password resets, 4 about billing, 4 about API rate limits, 2 singletons. Each gets an embedding at log time.

2. Find patterns — the analyzer loads all unresolved questions, ensures embeddings exist, builds a similarity matrix, and clusters them. You should see 3 patterns. The singletons get dropped.

3. Suggest documentation — pass "password reset" as a topic. The analyzer finds related questions by embedding similarity, then asks the LLM to generate a doc outline: title, sections, what to cover.

4. Mark resolved — mark the first two questions as resolved, verify the resolution rate updates.

5. Edge cases — resolve a nonexistent ID (returns False), run patterns on an empty database (returns []).

The output shows similarity scores between questions as they cluster, so you can see exactly which questions the algorithm considers similar and by how much. This is where you'll develop intuition for whether your threshold is right.

Things to know

The greedy algorithm is order-dependent. Shuffle the input and you might get slightly different clusters. For broad pattern detection — "people keep asking about refunds" — this is fine. For precise categorization, it's not. HDBSCAN would fix it.

text-embedding-3-large would tighten the clusters. Higher-dimensional embeddings produce sharper similarity scores, which means the threshold could go back up toward 0.65–0.70. I stuck with small because it's cheaper and the lower threshold compensates.

No auth. This server relies on the MCP client handling access control. If you're exposing it over HTTP instead of stdio, add authentication.

Wrapping Up

What we built: an MCP server that turns your RAG chatbot's silent failures into actionable insights. Log the questions it can't answer, cluster them by similarity, figure out what docs are missing, write them, close the loop.

The code is simple on purpose. SQLite, greedy clustering, 30 lines of core algorithm. The trade-offs are real, but the feedback loop works — and that's the part that matters.

Repo: github.com/hamurda/rag-insights-mcp

Fork it, plug it in, see what your users are actually asking that you haven't documented.

Have questions or feedback? Find me on LinkedIn or check out my multi-tenant AI chatbot series for the RAG chatbot this plugs into.

How I Scoped a RAG Knowledge Base Per Tenant

Damla Hamurcu — Mon, 09 Mar 2026 02:56:02 +0000

I'm building a multi-tenant AI chatbot, the kind where businesses embed a chat widget on their site, upload their docs, and get an AI assistant that actually knows their stuff.

Who is the best friend that comes everywhere with every multi-tenant system? Tenant Isolation! My chatbot was no exception.

How do I make sure Tenant A's documents never leak into Tenant B's responses?

This isn't a theoretical concern. If Company A uploads their internal pricing guide and Company B asks "what are your prices?", the vector similarity search doesn't care about organizational boundaries. It just finds the closest match. And that match might belong to someone else entirely.

The fix isn't complicated or fancy. But there are a few ways to do it, and I want to walk through what I picked and why.
Here's how I solved it using a single Supabase Postgres database, pgvector, and a tenant_id foreign key that does more heavy lifting than you'd expect.

The Problem: Vector Search Has No Concept of "Yours" vs "Mine"

RAG works by converting documents into chunks, embedding those chunks into vectors, and storing them. When a user asks a question, you embed the query, run a similarity search, pull the top matching chunks, and inject them into the LLM prompt as context. The user's prompt becomes like a history homework; here's the topic, here are the resources.

The trouble with multi-tenancy is that similarity search is global by default. If two tenants upload documents about "refund policy," the embeddings will be close in vector space regardless of which tenant owns them. Without explicit scoping, your retrieval step will happily mix them together. You asked about Yoda and it handed you Gandalf. Wrong universe entirely.

I needed isolation at two levels: the vector store (document chunks and embeddings) and the chat history (sessions and messages). Both contain tenant-specific data, and both needed to be locked down.

What I Considered (and Why I Rejected Most of It)

Option 1: Separate Database Per Tenant

The nuclear option. Every tenant gets their own Postgres instance with their own pgvector extension.

Sounds clean in theory, but it's a management nightmare. You're now maintaining N database connections, N migration pipelines, N backup schedules. On Supabase specifically, that means N separate projects with N separate billing lines.

Who on earth would need this? Enterprise customers or government projects with hard compliance requirements. But for a SaaS chatbot platform serving dozens or hundreds of small businesses? Absolutely not.

Option 2: Separate Tables Per Tenant

A bit less extreme. One database, but document_chunks_tenant_abc, document_chunks_tenant_def, and so on.

This is a theoretical option at best. Think about what happens at 50 tenants. Your migration scripts now need to create and alter 50 sets of tables. Your ORM layer needs to dynamically route to the right table name. Your vector indexes multiply. And when you inevitably need to add a column, you're running the same ALTER TABLE 50 times.

I can't think of a reason to prefer this over the other two.

Option 3: Shared Tables with `tenant_id` Filtering

One table for document chunks. One table for chat sessions. One table for messages. Every row has a tenant_id foreign key that points back to the tenants table. All queries filter by tenant_id.

This is what I went with. It's the simplest approach that actually scales, and combined with Supabase's Row Level Security policies, it provides genuine isolation, not just at the application layer, but at the database layer too.

The Data Model

Everything radiates outward from a central tenants table.

The Tenant

Each tenant represents a business using the platform. They get their own AI configuration: model, temperature, system prompt, welcome message:

class Tenant(Base):
    __tablename__ = "tenants"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
    name = Column(String(255), nullable=False)
    is_active = Column(Boolean, default=True, nullable=False)

    # LLM Configuration — each tenant can tune their own AI
    system_prompt = Column(Text, nullable=False,
        default="You are a helpful AI assistant.")
    model_name = Column(String(100), nullable=False, default="gemini-2.5-flash")
    temperature = Column(Numeric(3, 2), nullable=False, default=0.7)
    max_tokens = Column(Integer, nullable=False, default=1000)
    welcome_message = Column(Text, nullable=False,
        default="Hello! How can I help you today?")

    # Relationships — this is where tenant scoping starts
    documents = relationship("Document", back_populates="tenant",
        cascade="all, delete-orphan")
    chat_sessions = relationship("ChatSession", back_populates="tenant",
        cascade="all, delete-orphan", passive_deletes=True)

The key insight: every downstream relationship cascades from tenant_id. Delete a tenant, and their documents, chunks, embeddings, sessions, and messages all disappear. No orphaned vectors floating around.

Documents and Chunks

When a tenant uploads a PDF, it becomes a Document record. The processing pipeline then extracts text, splits it into chunks, generates embeddings, and stores everything in DocumentChunk:

class Document(Base):
    __tablename__ = "documents"

    id = Column(PGUUID(as_uuid=True), primary_key=True, default=uuid4)
    tenant_id = Column(PGUUID(as_uuid=True),
        ForeignKey("tenants.id", ondelete="CASCADE"),
        nullable=False, index=True)
    filename = Column(String(255), nullable=False)
    status = Column(String(50), nullable=False, default="pending")
    total_chunks = Column(Integer, default=0)


class DocumentChunk(Base):
    __tablename__ = "document_chunks"

    id = Column(PGUUID(as_uuid=True), primary_key=True, default=uuid4)
    document_id = Column(PGUUID(as_uuid=True),
        ForeignKey("documents.id", ondelete="CASCADE"),
        nullable=False, index=True)
    tenant_id = Column(PGUUID(as_uuid=True),
        ForeignKey("tenants.id", ondelete="CASCADE"),
        nullable=False, index=True)

    chunk_index = Column(Integer, nullable=False)
    content = Column(Text, nullable=False)
    token_count = Column(Integer, nullable=False)

    # The vector — 1536 dimensions from Gemini's embedding model
    embedding = Column(Vector(1536), nullable=True)

Notice that DocumentChunk has both document_id and tenant_id. The document_id tells you which file this chunk came from. The tenant_id is denormalized on purpose: it lets you filter by tenant directly on the chunks table without joining through documents. When you're running a cosine similarity search across potentially millions of vectors, that extra join is a cost you don't want to pay.

The vector index uses IVFFlat with cosine distance:

Index(
    "idx_chunks_embedding",
    "embedding",
    postgresql_using="ivfflat",
    postgresql_ops={"embedding": "vector_cosine_ops"}
)

IVFFlat is one of pgvector's indexing strategies. It clusters vectors into lists and searches the nearest clusters instead of scanning every row. Not the newest option (HNSW is generally better for growing datasets), but it's what I started with and it works.

Chat Sessions and Messages

Same pattern for the conversation side. Sessions belong to a tenant, messages belong to a session:

class ChatSession(Base):
    __tablename__ = "chat_sessions"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
    tenant_id = Column(UUID(as_uuid=True),
        ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False)
    anonymous_user_id = Column(UUID(as_uuid=True), nullable=False)
    status = Column(String(20), nullable=False)  # active, disconnected, ended


class ChatMessage(Base):
    __tablename__ = "chat_messages"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
    session_id = Column(UUID(as_uuid=True),
        ForeignKey("chat_sessions.id", ondelete="CASCADE"),
        nullable=False, index=True)
    role = Column(String(20), nullable=False)  # user, assistant, system
    content = Column(Text, nullable=False)
    tokens_used = Column(Integer, nullable=True)

Chat messages don't need their own tenant_id because they're always accessed through a session, which already has one. The scoping chain is: tenant → session → messages.

Where Isolation Meets the RAG Query

The schema sets the foundation, but isolation is enforced at query time. Here's the RAG retrieval service:

class RAGService:
    async def retrieve_context(
        self,
        query: str,
        tenant_id: UUID,      # This is the isolation boundary
        top_k: int = None,
        min_relevance: float = None,
    ) -> Tuple[str, List[Dict]]:

        # 1. Embed the user's question
        query_embedding = await self.embedding_service.embed_query(query)

        # 2. Semantic search — scoped to this tenant only
        results = await self.chunk_repo.semantic_search(
            query_embedding=query_embedding,
            tenant_id=tenant_id,       # RIGHT HERE
            top_k=top_k,
            min_similarity=min_relevance,
        )

        # 3. Build context for the LLM
        context = self._build_context(results)
        citations = self._build_citations(results)
        return context, citations

The tenant_id parameter is non-optional. There's no code path where a similarity search runs without it. That was a deliberate choice, not an accident. I wanted it to be impossible to forget. The underlying SQL looks roughly like this:

SELECT dc.*, 1 - (dc.embedding <=> :query_embedding) AS similarity
FROM document_chunks dc
WHERE dc.tenant_id = :tenant_id
  AND 1 - (dc.embedding <=> :query_embedding) >= :min_similarity
ORDER BY dc.embedding <=> :query_embedding
LIMIT :top_k;

The <=> operator is pgvector's cosine distance. The WHERE dc.tenant_id = :tenant_id clause means Postgres only considers vectors belonging to this specific tenant. It doesn't matter that another tenant uploaded a nearly identical document — their chunks are filtered out before the distance calculation even runs.

The Database-Level Safety Net: Row Level Security

Application-level filtering is necessary but not sufficient. What if a bug skips the tenant_id filter? What if a future developer writes a query without it? That's where Supabase's Row Level Security (RLS) comes in.

RLS policies run at the database engine level. Even if your application code forgets the filter, the database won't return rows the current session shouldn't see. Your code can have a bad day. The database still has your back.

The setup involves setting a session-level config variable with the current tenant's ID, then writing policies that check it:

-- Enable RLS on sensitive tables
ALTER TABLE document_chunks ENABLE ROW LEVEL SECURITY;
ALTER TABLE documents ENABLE ROW LEVEL SECURITY;
ALTER TABLE chat_sessions ENABLE ROW LEVEL SECURITY;

-- Policy: only return rows matching the session's tenant
CREATE POLICY "tenant_isolation" ON document_chunks
    FOR ALL
    USING (tenant_id::text = current_setting('app.tenant_id', true));

On the application side, every request sets this config before executing queries:

async def get_tenant_db_session(
    current_tenant_id: str = Depends(get_current_tenant_id),
    db: AsyncSession = Depends(get_async_session)
) -> AsyncGenerator[AsyncSession, None]:
    # Set tenant context at the database session level
    await db.execute(
        text("SELECT set_config('app.tenant_id', :tenant_id, false)"),
        {"tenant_id": current_tenant_id}
    )
    await db.commit()

    yield db

The false in set_config matters. It means the tenant context persists for the entire database session, not just the current transaction. Change that to true and your subsequent queries in the same session lose the filter.

The tenant_id itself comes from a JWT token that's issued when a chat widget initializes with a valid API key. The chain is: API key → tenant lookup → JWT with tenant_id → session config → RLS enforcement. There's no point in the chain where a user can substitute a different tenant's ID.

How It All Flows Together

Here's the full lifecycle of a single chat message:

Widget loads on a business's website. It sends the publishable API key (pk_live_...) to initialize a session.
Backend validates the key, identifies the tenant, creates a chat session, and returns a JWT containing the tenant_id.
User sends a message over WebSocket. The handler extracts the tenant from the JWT.
RAG retrieval embeds the question and searches document_chunks, filtered by this tenant's ID. Only this tenant's documents are considered.
Context injection — the matching chunks are formatted and appended to the system prompt, which itself is built from this tenant's configuration.
LLM response is streamed back through the WebSocket. Citations linking the response to specific document chunks are saved.
Everything persists under this tenant's ID: the message, the citations, the token counts.

At no point does the system touch another tenant's data. The application layer filters by tenant_id. The database layer enforces it through RLS. The authentication layer guarantees the tenant_id is legitimate.

What I'd Do Differently Next Time

The architecture works. I haven't hit scaling limits yet, but here's what I'm watching for:

Partition the chunks table. Right now all tenants share one document_chunks table and one IVFFlat index. As the table grows past a few million rows, performance could degrade. Postgres table partitioning by tenant_id (range or list partitioning) would let each tenant's chunks live in a physical partition with its own vector index, while still presenting as one logical table to the application layer.

Add embedding model versioning. If you switch embedding models (say, from text-embedding-ada-002 to gemini-embedding-001), old chunks become incompatible with new query embeddings. A model_version column on chunks would let you handle migrations gracefully.

The Takeaway

Multi-tenant RAG doesn't require exotic infrastructure. A single Postgres database with pgvector, a disciplined tenant_id on every table that holds tenant data, and Row Level Security as a safety net gives you genuine isolation without the operational overhead of running separate databases.

The architecture is three things working together: schema design (every row knows its tenant), application logic (every query filters by tenant), and database policy (RLS catches anything the application misses). Remove any one of these and you have a gap. Keep all three and you have a system where tenant data leakage isn't just unlikely — it's structurally prevented.

This is Part 3 in a series on building a multi-tenant AI chatbot platform.
Part 1 covered How I Solved WebSocket Authentication in FastAPI.
Part 2 covered Why My Multi-Tenant Chatbot Needed Two Types of API Keys.
The stack is Python/FastAPI, Supabase (Postgres + pgvector), and Gemini/OpenAI for LLM and embeddings.

Why My Multi-Tenant Chatbot Needed Two Types of API Keys

Damla Hamurcu — Mon, 02 Mar 2026 05:29:08 +0000

I'm building a multi-tenant AI chatbot. Businesses sign up, share their documents and get a chat widget to embed on their website. The widget talks to their knowledge base and only their knowledge base. I covered the WebSocket authentication side of this in a previous post. This one is about the problem that lives on the other end — in the browser.

The widget needs to know which tenant it belongs to. That means some form of credential has to exist in the frontend code. As we all know, frontend code is not a friend who can keep secrets. The code is public. Anyone can open DevTools, inspect the network tab, and read the script tag. Whatever key you put there is fully exposed from day one.

So the question becomes: how do you authenticate something that can never be secret?

Where Do You Even Store This Stuff?

Now I need to make a tiny side track here. Before I could even get to the API key problem, I had a session management problem. When an anonymous user opens the chat widget, the backend creates a session and returns three things: a session_id, an anonymous_user_id, and a JWT token. The widget needs to hold onto these so that if the user closes the tab and comes back twenty minutes later, they can pick up where they left off instead of starting a blank conversation.

That ruled out sessionStorage immediately — it dies when the tab closes. I needed localStorage.

But localStorage is scoped per origin, not per page. If a user happens to visit two different websites that both embed my chatbot widget (unlikely, but not impossible), the widget JavaScript runs on their respective domains. That's fine — different origins, different storage. The real edge case is during development and testing, or if the widget is loaded from a shared domain. I didn't want to leave that door open.

The fix was to namespace the storage keys using a portion of the API key. The publishable key is already in the widget's script tag — it's the one identifier that's guaranteed to be unique per tenant and available before any API call happens. So the storage looks like this:

// API key from the script tag: pk_live_a8Kx9mNq...
const prefix = apiKey.slice(8, 14); // "a8Kx9m"

localStorage.setItem(`${prefix}_session_id`, sessionId);
localStorage.setItem(`${prefix}_anonymous_user_id`, visitorId);
localStorage.setItem(`${prefix}_token`, jwt);

Now two different tenants' widgets on the same origin write to different keys. No collisions, no cross-contamination.

When the widget loads, it checks for existing keys under its prefix. If it finds them, it sends them to the backend to resume the session. If the session has expired, the widget creates a new one but reuses the same anonymous_user_id so the tenant can still recognize a returning visitor.

The Key That Does Too Much

At this point I had one API key per tenant. It authenticated the widget, identified the tenant, and in theory would also authenticate the admin dashboard I was planning. You know… The interface where tenants manage their uploaded documents, update their system prompt, tweak their model settings.

Hopefully this is where it hits us if not earlier. The key in the widget's script tag is permanently exposed. It's not a leak, it's by design! But if that same key can also delete documents or change the AI's behavior, then anyone who reads the page source can do those things too.

One way to limit this could be managing allowed origins accessing the endpoints. It sounds more like a band-aid solution, or a helper at best. Cannot be the main solution for sure; origins can be spoofed, and it doesn't help if the key is used from a tool like Postman or curl.

I needed the widget key to be able to create chat sessions and send messages. Nothing more. And I needed a separate key for administrative operations that would never appear in frontend code. One that's shown to the tenant exactly once when it's generated and never transmitted over a public channel again.

Two different security contexts. Two different keys.

Splitting Publishable and Secret Keys

So… I landed on a naming convention borrowed from online examples: pk_live_ for publishable keys and sk_live_ for secret keys.

The publishable key (pk_live_) goes in the widget. It's public by nature. Its scopes are limited to sessions:create and sessions:read. It can initialize a chat session, refresh a JWT, and resume a conversation. It cannot touch documents, settings, or anything administrative.

The secret key (sk_live_) is for the admin dashboard and backend integrations. It's shown once at creation time. Until I build the admin UI with proper login, the tenant copies it, stores it somewhere secure, and it never appears in a browser. It carries the admin scope, which grants access to document uploads, deletions, configuration changes, and later things like OAuth connections for external integrations.

Both key types live in a dedicated api_keys table:

class APIKey(Base):
    __tablename__ = "api_keys"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
    tenant_id = Column(UUID(as_uuid=True),
        ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False)
    key_hash = Column(String(64), nullable=False)    # SHA-256, never store raw
    key_type = Column(String(20), nullable=False)     # "publishable" or "secret"
    scopes = Column(JSON, nullable=False, default=list)
    is_active = Column(Boolean, default=True, nullable=False)
    created_at = Column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))
    last_used_at = Column(DateTime(timezone=True), nullable=True)
    revoked_at = Column(DateTime(timezone=True), nullable=True)

I deliberately made this a separate table rather than adding key columns to the tenant table. A tenant can have multiple keys — maybe an old publishable key that's still in a cached version of someone's website, plus the current one, plus a secret key that was rotated after an employee left. Each key has its own lifecycle: creation time, last used time, revocation time. Putting that on the tenant table would mean either a mess of nullable columns or losing the ability to track key history entirely.

The raw key is only ever returned once, at generation time. What gets stored is a SHA-256 hash:

async def generate_for_tenant(self, tenant_id, key_type, scopes):
    if key_type == APIKeyType.PUBLISHABLE:
        raw_key = generate_api_key(prefix="pk_live")
    else:
        raw_key = generate_api_key(prefix="sk_live")

    key_hash = hashlib.sha256(raw_key.encode()).hexdigest()

    await self.repo.create(
        tenant_id=tenant_id,
        key_hash=key_hash,
        key_type=key_type.value,
        scopes=scopes
    )

    return raw_key  # Only time this is returned

When a request comes in, the provided key is hashed and looked up. If the hash matches an active, non-revoked key, the request is authenticated and the tenant is identified. If the key is compromised, you deactivate it and generate a new one. The tenant record is untouched — only the api_keys table changes.

Validation checks everything in sequence: does the hash exist, is the key active, is it not revoked, is the tenant active, does it have the required scopes (for simplicity we'll return None in this snippet):

async def validate_key(self, raw_key, required_scopes=None):
    key_hash = hash_api_key(raw_key)
    api_key = await self.repo.get_by_hash_with_tenant(key_hash)

    if not api_key:              return None
    if not api_key.is_active:    return None
    if api_key.revoked_at:       return None
    if not api_key.tenant.is_active: return None

    if required_scopes:
        if not api_key.has_any_scope(required_scopes):
            return None

    return {
        "tenant_id": str(api_key.tenant_id),
        "tenant": api_key.tenant,
        "key_type": api_key.key_type,
        "scopes": api_key.scopes
    }

In production, each of those return None cases logs a different reason — expired key, revoked key, inactive tenant — so I can debug authentication failures without exposing details to the caller.

Two Apps, One Backend

The key split had a natural downstream effect on the application structure. The widget and the admin dashboard have different authentication mechanisms, different CORS policies, and different route sets. Cramming them into one FastAPI app with a bunch of conditional middleware felt wrong.

Instead, I split them into two sub-applications mounted under the same FastAPI instance:

# main.py
app = FastAPI(title="RAG Chatbot API")

app.mount("/widget", widget_app)
app.mount("/admin", admin_app)

The widget app allows only specific origins (the tenant's website), exposes only X-Chatbot-Key in its CORS headers, and serves two routes: session initialization and WebSocket connections. Its authentication dependency reads the pk_live_ key from the X-Chatbot-Key header:

# widget_app.py
widget_app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.widget_cors_origins_list,
    allow_credentials=False,
    allow_methods=["GET", "POST", "OPTIONS"],
    allow_headers=["Content-Type", "X-Chatbot-Key"],
)

The admin app has wider CORS allowances (for the dashboard frontend), uses X-API-Key with sk_live_ keys, and will eventually serve document management, tenant configuration, and OAuth connection routes.

This separation also made the WebSocket flow cleaner. The widget app's WebSocket endpoint delegates to a SessionManager that orchestrates the entire connection lifecycle:

context = await session_manager.establish_session(websocket, token)
if not context:
    return  # Connection already closed with reason

# context now holds: session_id, tenant_id, tenant config, websocket
# Ready to process messages

Under the hood, establish_session verifies the JWT, validates the session, loads the tenant config, and registers the connection. Each step can fail independently and close the connection with a specific WebSocket close code. Without the sub-app boundary making ownership clear, this orchestration logic would have been tangled up with admin concerns.

What I'd Do Differently

Rate limiting per publishable key from day one. Right now, rate limiting happens per session. But a bad actor with the publishable key could create sessions in a loop. I'd add a rate limit at the key level — say, 100 session creations per hour per publishable key — before it becomes a problem.

A tighter localStorage cleanup strategy. The namespaced keys work, but old sessions accumulate. I'd add a TTL check on widget initialization that clears stale entries — anything with an expired token older than 24 hours.

Consider key rotation automation. Right now, key rotation is manual: revoke the old one, generate a new one, update the widget embed. For enterprise tenants, an automated rotation flow with a grace period (both old and new keys active for 24 hours) would reduce friction.

Recap

Here's the chain of decisions, each one causing the next:

The widget needed to survive tab closes → localStorage, not sessionStorage.
Two tenants on the same origin would collide → namespace storage keys with the API key prefix.
The API key is permanently public in the widget → it shouldn't have admin permissions.
You need admin operations too → create a second key type with separate scopes.
Two key types need lifecycle tracking → dedicated api_keys table, not columns on the tenant.
Two authentication contexts → two sub-apps with isolated middleware and routes.

None of this was planned upfront. Each decision was forced by the one before it. That's usually a sign you're solving real problems rather than imaginary ones.

If you've hit similar decisions building multi-tenant systems or solved them differently, I'd love to hear about it in the comments.

This is Part 2 in a series on building a multi-tenant AI chatbot platform. Part 1 covered How I Solved WebSocket Authentication in FastAPI (And Why Depends() Wasn't Enough). The stack is Python/FastAPI, Supabase (Postgres + pgvector), and Gemini/OpenAI for LLM and embeddings.

I Turned My Piano Practice Into a Fantasy Adventure (Because I Kept Skipping It)

Damla Hamurcu — Sat, 28 Feb 2026 04:28:06 +0000

This is a submission for the DEV Weekend Challenge: Community

The Community

Adult piano learners are a surprisingly large and quietly frustrated community. We're not children being marched to the piano by parents. We chose this. We pay for lessons, we have the instrument, we genuinely want to play — and yet, week after week, we sit down to practice and feel... nothing. A list of tasks. Bars 12 to 20, hands separate. C major scale, two octaves. Again.

The problem isn't motivation in the grand sense. We love music. The problem is that practice, stripped of narrative, feels like admin. And admin is easy to skip.

I've been learning piano as an adult for a while now. My teacher gives me a list of tasks each week. They're good tasks. I just rarely feel excited to open the list. There's no story. No stakes. No sense that anything meaningful is happening between one Tuesday and the next.

I built Practice Quest for this community: adult learners who are technically capable of practicing but emotionally a bit resistant to it.

What I Built

Practice Quest is a web app that transforms your weekly piano practice tasks into a fantasy quest narrative, powered by AI.

You open the app, type in your practice tasks for the week (whatever your teacher assigned, or whatever you're working on) and hit "Begin the Journey." The app sends each task to OpenAI, which writes a unique atmospheric quest description for it: a location, a title, a 2-3 sentence story that frames the task as something meaningful and adventurous.

Your scales become a trial at The Crystal Cavern. Your sight-reading becomes an ancient manuscript discovered in The Mossy Archive. Bars 12-20 of your piece become the key to unlocking The Amber Gate.

You work through the quests one at a time. Each completion triggers a small sparkle animation and unlocks the next. When you finish everything, the journey is complete and the realm is at peace.

It's a small reframe. But that reframe is the whole point, practice should feel like progress, not admin.

Demo

practice-quest.vercel.app

Code

hamurda / practice-quest

Takes your weekly piano tasks and weaves them into an atmospheric quest narrative, powered by OpenAI.

Practice Quest 🧭

Transform your piano practice into a fantasy adventure.

Practice Quest takes your weekly piano tasks and weaves them into an atmospheric quest narrative — powered by OpenAI. Each task becomes a location, a trial, a story. Completing practice feels like progressing through a world, not ticking a checkbox.

Built for adult piano learners who know the struggle of making themselves sit down and practice.

Deploy to Vercel

1. Clone this repo

git clone https://github.com/yourusername/practice-quest
cd practice-quest

2. Install Vercel CLI

npm install -g vercel

3. Deploy

vercel

Follow the prompts — select "no" for existing project, accept defaults.

4. Add your OpenAI API key

In the Vercel dashboard → your project → Settings → Environment Variables:

OPENAI_API_KEY = sk-...your key here

Then redeploy:

vercel --prod

That's it. Your app is live.

Project Structure

practice-quest/
├── index.html        # The entire frontend
├── api/
│   └── narrate.js    # Vercel

…

View on GitHub

The project is deliberately minimal: one HTML file for the entire frontend, one Vercel serverless function to proxy the OpenAI call, and a vercel.json. No framework, no build step, no database.

How I Built It

The stack is intentionally small:

Vanilla HTML/CSS/JS for the frontend, single file, no framework. The parchment aesthetic, animations, and quest card layout are all hand-written CSS.
OpenAI gpt-4.1-mini for the narration. Each task gets its own API call with a system prompt that instructs the model to write atmospheric, encouraging quest text referencing the fantasy location, ending with a sense of anticipation, and making the practice feel heroic rather than tedious.
Vercel serverless function (/api/narrate.js) as a thin proxy, this exists entirely to keep the OpenAI API key off the client. The frontend calls /api/narrate, the function calls OpenAI, the key never touches the browser.
Graceful fallback, if the API call fails for any reason, the app silently falls back to template-generated flavour text. Nothing breaks for the user.

The loading screen (an animated quill writing, with per-task progress indicators) is probably my favourite part. It gives the AI generation a sense of ritual like a scribe is actually composing your adventure.

On the design: I wanted it to feel like a parchment map, not an app. Crimson Text for the serif, Quicksand for UI labels, a warm parchment background with subtle noise texture, gold as the primary accent. The quest cards animate in on a slight delay when the journey loads. Completing a quest triggers gold sparkles from the button.

Why this community specifically: I'm in it. I know the exact moment of friction, opening the practice notebook, seeing a list of technical tasks, feeling the fun drain out before you've played a single note. The hypothesis Practice Quest tests is whether reframing that list as a quest changes how the session feels. For me, this week at least, it did.

How I Solved WebSocket Authentication in FastAPI (And Why Depends() Wasn't Enough)

Damla Hamurcu — Mon, 23 Feb 2026 04:16:17 +0000

I'm a developer transitioning from enterprise systems to building my own products. This is one of the real problems I solved along the way.

The Problem

I was building a multi-tenant RAG chatbot. One backend serving AI chat widgets across multiple client websites. Each widget needed to identify itself to the backend, and I assumed I'd handle that the way I always had: with a header.

The stack was simple: FastAPI backend and a Vite frontend widget. That's it. I also wanted real-time streaming so users could see the AI response being generated word by word and stop it mid-stream if needed. Like we all do with ChatGPT. That meant WebSocket connections instead of plain REST APIs.

In order to manage chat history, knowledge base, leads and automations, I needed three pieces of information from the start: tenant id to keep the bot in personality and constrain its knowledge base, session id to track the chat session, and anonymous user id to identify returning users. That, my friends, is what I'll be calling tenant context from now on.

As you'd expect, establishing tenant context starts with validating the tenant. The original plan was straightforward — send the API key in a custom header, capture it with FastAPI's Header() and Depends(), let the backend do its magic and establish the connection.

However. Browser's WebSocket API doesn't let you set custom headers. If you were connecting server-to-server, headers would work fine. But from a browser widget? No-go.

So there I was. On one hand, I wanted to stick with WebSocket connections to achieve the streaming experience. On the other hand... security.

The Naive Approach

The most obvious workaround is using a query parameter instead of a header:

ws://api.example.com/chat?api_key=pk_live_abc123

This works. But look at what's sitting in that URL. Your API key, right there in server logs, browser history, and every proxy between the client and your backend. That's a security problem.

And it gets worse. The WebSocket handshake is an HTTP GET upgrade request. There's no request body. So you can't sneak credentials in the body either. Query params are basically your only option for passing data during the handshake.

Even if I accepted that risk, there was a second problem: timing.

By the time my WebSocket endpoint received the connection, the client was already in. The connection was established. But my backend still needed to validate the API key, look up the tenant, set the session, and build the full context. All of that was happening after the handshake.

The result? A chatbot that looked broken. The widget would connect, but the user couldn't actually chat for a few seconds while the backend scrambled to set things up behind the scenes. Not a great first impression.

There's also a real security concern here. Once you call await websocket.accept(), that client has an open connection consuming server resources like memory, a file descriptor, a slot in your connection manager. If you accept first and validate later, anyone can open thousands of connections with garbage credentials and exhaust your server. Classic denial-of-service vector.

It became clear pretty quickly that trying to do everything — authentication, context setup, and chat — within a single WebSocket connection was limiting and convoluted.

The "Aha" Moment

I split the process into two separate API calls.

Step 1: A plain REST POST request. The widget sends the API key in the header (where it belongs). The backend validates the tenant, creates or refreshes the session, and returns a short-lived JWT token — good for maybe 5 minutes, just long enough to establish the WebSocket connection.

Step 2: The widget opens a WebSocket connection with that token in the query.

POST https://api.example.com/session   →  returns token
 WS  ws://api.example.com/chat?token=eyJhbGciOi…  →  establishes connection

Now the API key never appears in the WebSocket URL. The token in the query is short-lived, scoped to a specific session, and doesn't expose the actual credential. Even if someone intercepts it, it expires in minutes.

And here's a side benefit I didn't expect. While the frontend waits for the token from that first REST call, the UI can show a "Connecting..." indicator. Then when the WebSocket opens, the bot is immediately ready to chat, no awkward delay. The perceived latency actually dropped compared to the single-connection approach.

Two calls. More secure. Feels faster. I'll take it.

The Solution, Step by Step

Let me walk you through the actual implementation. I'm showing simplified versions here to keep the focus on the pattern. In production, you'll want to add UUID validation, session expiry handling, and tenant ownership checks on top of this.

Here's the flow:

Widget                          Backend
  │                                │
  │── POST /session ──────────────>│  (API key in header)
  │   Body: { session_id?, user_id? }
  │<── { token, session_id } ──────│
  │                                │
  │── WS /chat?token=eyJhbG... ──>│  (validate token before accept)
  │<════════ connected ════════════│
  │                                │
  │── { message: "hello" } ──────>│
  │<── { chunk: "Hi! How can..." } │  (streaming response)

Step 1: The REST Endpoint

When the widget loads, it makes a POST request to create or refresh a session.

@router.post("/session")
async def create_or_refresh_session(
    request: SessionCreateRequest,
    tenant: Tenant = Depends(get_tenant_from_publishable_key),
    session_service: ChatSessionService = Depends(get_chat_session_service),
):
    if request.session_id:
        # Returning user — refresh the existing session
        result = await session_service.refresh_session(
            session_id=UUID(request.session_id),
            tenant_id=tenant.id,
        )
    else:
        # New visitor — create a fresh session
        result = await session_service.create_session(
            tenant_id=tenant.id,
            anonymous_user_id=(
                UUID(request.anonymous_user_id) 
                if request.anonymous_user_id else None
            ),
        )
    return SessionCreateResponse(**result)  # includes the JWT token

Notice what's happening here. Depends(get_tenant_from_publishable_key) does the heavy lifting. It pulls the API key from the header, validates it against the database, and returns the tenant. Classic FastAPI dependency injection, clean and testable.

Here's what that dependency looks like under the hood:

async def get_publishable_key_from_header(
    x_api_key: Annotated[str | None, Header(alias="api-key")] = None,
) -> str:
    if not x_api_key:
        raise HTTPException(
            status_code=401,
            detail="Missing API key. Include api-key header.",
        )
    return x_api_key


async def get_tenant_from_publishable_key(
    api_key: str = Depends(get_publishable_key_from_header),
    tenant_service: TenantService = Depends(get_tenant_service),
) -> Tenant:
    tenant = await tenant_service.validate_key(api_key)
    if not tenant:
        raise HTTPException(status_code=401, detail="Invalid API key")
    return tenant

The response includes a JWT containing the session id, tenant id, and anonymous user id — everything the WebSocket endpoint needs to establish context without another database lookup.

Step 2: The WebSocket Endpoint

Once the frontend has the token, it opens the WebSocket connection:

@router.websocket("/ws")
async def websocket_endpoint(
    websocket: WebSocket,
    token: str = Query(..., description="JWT from POST /session"),
    session_service: ChatSessionService = Depends(get_chat_session_service),
    connection_manager: ConnectionManager = Depends(get_connection_manager),
):
    # Validate BEFORE accepting the connection
    session_context = validate_and_decode_token(token)

    if not session_context:
        await websocket.close(code=4001, reason="Invalid or expired token")
        return

    # Token is valid — now we accept
    await websocket.accept()

    try:
        while True:
            data = await websocket.receive_json()
            # Handle messages — all business logic delegated to handlers
            await handle_message(data, session_context, websocket)

    except WebSocketDisconnect:
        logger.info(f"Disconnected: {session_context.session_id}")

    finally:
        await cleanup_session(session_context.session_id, websocket)

The key line is right at the top. validate_and_decode_token(token) runs before websocket.accept(). If the token is invalid or expired, we close the connection immediately. The client never gets in. No wasted resources. No DoS vector.

Here's what the token validation looks like:

def validate_and_decode_token(token: str) -> SessionContext | None:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return SessionContext(
            session_id=payload["session_id"],
            tenant_id=payload["tenant_id"],
            anonymous_user_id=payload.get("anonymous_user_id"),
        )
    except jwt.ExpiredSignatureError:
        return None
    except jwt.InvalidTokenError:
        return None

Simple. Decode the JWT, extract the context, return it. If anything is wrong, return None and the WebSocket endpoint handles the rejection.

What the Frontend Does

For completeness, here's what the widget side looks like in essence:

// Step 1: Get the session token
const res = await fetch("https://api.example.com/session", {
  method: "POST",
  headers: {
    "Content-Type": "application/json",
    "api-key": "pk_live_abc123",  // safe in a header
  },
  body: JSON.stringify({
    session_id: existingSessionId || undefined,
    anonymous_user_id: visitorId || undefined,
  }),
});
const { token } = await res.json();

// Step 2: Connect WebSocket with the token
const ws = new WebSocket(
  `wss://api.example.com/ws?token=${token}`
);

ws.onopen = () => {
  console.log("Connected and ready to chat");
};

Two requests. First one authenticates and sets up the session. Second one connects. By the time the WebSocket opens, the backend already knows exactly who it's talking to.

Why This Pattern Matters Beyond My Project

When I first split authentication into two calls, it felt like I was overcomplicating things. Two requests where one should work? Surely there's a simpler way.

There isn't. And once you see why, it becomes obvious.

This is just separation of concerns applied to WebSocket authentication. REST does what REST is good at: request-response with full HTTP features like headers, status codes, and standard error handling. WebSocket does what it's good at: persistent bidirectional communication. Each protocol handles the part it was designed for.

Once you see it that way, putting authentication inside the WebSocket connection feels like using a screwdriver to hammer a nail. It technically works. But you're fighting the tool.

This pattern also gives you reconnection for free. WebSocket connections drop. They just do. Mobile networks, laptop sleep, flaky Wi-Fi. When that happens, the widget calls the REST endpoint again, gets a fresh token, and reconnects. The user sees "Reconnecting..." for a second and then they're back. No re-authentication headache.

After I built this, I started noticing the same two-step pattern described in WebSocket authentication discussions everywhere. Turns out it's the standard approach. I just had to arrive at it the hard way.

What I'd Do Differently

If I could go back and talk to past-me at the start of this project, I'd say two things.

First: stop trying to do everything inside the WebSocket connection. I kept pushing authentication, context setup, and session management into the handshake because that's what felt "right" coming from enterprise apps where one connection means one authentication flow. It introduced delay, complexity, and a chatbot that looked broken on load. The moment I unlearned that assumption and just used two calls, everything clicked.

Second: think harder about token duration. I set it to 5 minutes and moved on. But this choice matters more than I initially gave it credit for. Too long, and you've essentially recreated the API-key-in-the-query problem — a token sitting in the URL for minutes is an unnecessary exposure window. Too short, and users on slow networks can't establish the WebSocket connection before the token expires. There's a sweet spot and it depends on your use case. Test it. Don't just pick a number and forget about it like I almost did.

Quick Recap

Here's the pattern in short:

Browser WebSocket API doesn't support custom headers. Don't fight it.
Don't put API keys in query params. They end up in logs, browser history, and proxies.
Split authentication into two steps. REST call first to validate and issue a short-lived JWT. WebSocket connection second, with that token.
Validate the token before accepting the WebSocket connection. This prevents unauthorized connections from consuming server resources.
Use the REST endpoint for reconnection too. When the WebSocket drops, the widget gets a fresh token and reconnects seamlessly.

Remember, it's two calls because that's the right way to do it.

If you've hit this same wall or solved it differently, I'd love to hear about it in the comments.

DEV Community: Damla Hamurcu

I Let an LLM Play Cookie Clicker. Then I Fixed Its Biggest Weakness.

The Setup

Results

What Actually Happened

Why Greedy Is Secretly Strong

The LLM Doesn't Even Agree With Itself

The Control Loop Matters More Than the Intelligence

What This Means Beyond Cookie Clicker

What I'd Do Next

Try It Yourself

What Cookie Clicker Taught Me About AI-Assisted Development

The Project

Session 1 & 2: Spec-First Works — Until You Refactor

Session 3: Where Spec-First Hit Its Limits

Using the Tool to Check the Tool

The Patterns

The Judgment Layer

Build an MCP Server That Finds Your RAG Chatbot's Blind Spots

Setup

Step 1: Store a Question

Step 2: Add Embeddings

Step 3: Cluster by Similarity

About that threshold

Step 4: Complete the analysis pipeline

Step 5: Wrap It in MCP

Step 6: The Full Loop

Things to know

Wrapping Up

How I Scoped a RAG Knowledge Base Per Tenant

The Problem: Vector Search Has No Concept of "Yours" vs "Mine"

What I Considered (and Why I Rejected Most of It)

Option 1: Separate Database Per Tenant

Option 2: Separate Tables Per Tenant

Option 3: Shared Tables with tenant_id Filtering

The Data Model

The Tenant

Documents and Chunks

Chat Sessions and Messages

Where Isolation Meets the RAG Query

The Database-Level Safety Net: Row Level Security

How It All Flows Together

What I'd Do Differently Next Time

The Takeaway

Why My Multi-Tenant Chatbot Needed Two Types of API Keys

Where Do You Even Store This Stuff?

The Key That Does Too Much

Splitting Publishable and Secret Keys

Two Apps, One Backend

What I'd Do Differently

Recap

I Turned My Piano Practice Into a Fantasy Adventure (Because I Kept Skipping It)

The Community

What I Built

Demo

Code

hamurda / practice-quest

Takes your weekly piano tasks and weaves them into an atmospheric quest narrative, powered by OpenAI.

Practice Quest 🧭

Deploy to Vercel

Project Structure

How I Built It

How I Solved WebSocket Authentication in FastAPI (And Why Depends() Wasn't Enough)

The Problem

The Naive Approach

The "Aha" Moment

The Solution, Step by Step

Step 1: The REST Endpoint

Step 2: The WebSocket Endpoint

What the Frontend Does

Why This Pattern Matters Beyond My Project

What I'd Do Differently

Quick Recap

Option 3: Shared Tables with `tenant_id` Filtering