The night I gave up on an off-the-shelf bot
My first support bot sat on top of a hosted service. It worked, but when I wanted to change a single phrase in its replies, there was nowhere to put my hands. One evening I spent half an hour opening and closing admin tabs looking for that phrase, and I stopped.
Looking back, what I wanted wasn't a more capable bot. It was a bot I could open up. Once I rebuilt it directly on the Claude API, every character of the system prompt, every decision about where to trim history, and every choice about which model to pay for became mine to make.
What follows is the order I actually went in — from a minimal script up through history, streaming, a web API, cost, retries, and PII, one working rung at a time.
STEP 1: The Minimal Chatbot
Start with the simplest possible implementation.
import anthropic
client = anthropic.Anthropic(api_key="YOUR_API_KEY")
def chat(user_message: str) -> str:
message = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[
{"role": "user", "content": user_message}
]
)
return message.content[0].text
# Run it
response = chat("Write a Python function to check if a number is prime")
print(response)This works, but there's a critical flaw: conversation context isn't preserved. Each message starts fresh with no memory of previous turns.
STEP 2: Managing Conversation History
The Claude API is stateless, so you must manage conversation history on the client side.
import anthropic
from typing import List
client = anthropic.Anthropic(api_key="YOUR_API_KEY")
class ChatSession:
def __init__(self, system_prompt: str = ""):
self.history: List[dict] = []
self.system_prompt = system_prompt
def send(self, user_message: str) -> str:
# Add user message to history
self.history.append({
"role": "user",
"content": user_message
})
# Call API with full history
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=2048,
system=self.system_prompt,
messages=self.history
)
assistant_message = response.content[0].text
# Add assistant response to history
self.history.append({
"role": "assistant",
"content": assistant_message
})
return assistant_message
def clear(self):
"""Reset conversation"""
self.history = []
# Usage example
session = ChatSession(
system_prompt="You are a Python code reviewer. "
"Identify issues and suggest improvements with clear explanations."
)
print(session.send("Review this code:\ndef add(a, b): return a+b"))
print(session.send("Now add type hints to the improved version")) # Remembers contextSTEP 3: Implementing Streaming Responses
Long responses create a poor UX when users wait for the full reply. Stream tokens as they're generated.
import anthropic
client = anthropic.Anthropic(api_key="YOUR_API_KEY")
class StreamingChatSession:
def __init__(self, system_prompt: str = ""):
self.history = []
self.system_prompt = system_prompt
def send_stream(self, user_message: str):
"""Yield text chunks as they arrive"""
self.history.append({"role": "user", "content": user_message})
full_response = ""
with client.messages.stream(
model="claude-sonnet-4-6",
max_tokens=2048,
system=self.system_prompt,
messages=self.history
) as stream:
for text in stream.text_stream:
full_response += text
yield text # Return text incrementally
# Save complete response to history
self.history.append({"role": "assistant", "content": full_response})
# CLI example
session = StreamingChatSession(system_prompt="You are a helpful assistant.")
while True:
user_input = input("\nYou: ").strip()
if user_input.lower() in ["quit", "exit"]:
break
print("Claude: ", end="", flush=True)
for chunk in session.send_stream(user_input):
print(chunk, end="", flush=True)
print() # newlineSTEP 4: Exposing as a Web API with FastAPI
Go beyond CLI and build a proper API your frontend can consume.
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import anthropic
import json
from typing import Optional
app = FastAPI()
client = anthropic.Anthropic(api_key="YOUR_API_KEY")
# Simple in-memory session store (use Redis in production)
sessions: dict[str, list] = {}
class ChatRequest(BaseModel):
session_id: str
message: str
system_prompt: Optional[str] = "You are a helpful assistant."
class ChatResponse(BaseModel):
response: str
session_id: str
@app.post("/chat")
async def chat(request: ChatRequest):
if request.session_id not in sessions:
sessions[request.session_id] = []
history = sessions[request.session_id]
history.append({"role": "user", "content": request.message})
try:
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=2048,
system=request.system_prompt,
messages=history
)
assistant_message = response.content[0].text
history.append({"role": "assistant", "content": assistant_message})
return ChatResponse(response=assistant_message, session_id=request.session_id)
except anthropic.APIError as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
"""Server-Sent Events streaming endpoint"""
if request.session_id not in sessions:
sessions[request.session_id] = []
history = sessions[request.session_id]
history.append({"role": "user", "content": request.message})
def generate():
full_response = ""
with client.messages.stream(
model="claude-sonnet-4-6",
max_tokens=2048,
system=request.system_prompt,
messages=history
) as stream:
for text in stream.text_stream:
full_response += text
yield f"data: {json.dumps({'text': text})}\n\n"
history.append({"role": "assistant", "content": full_response})
yield f"data: {json.dumps({'done': True})}\n\n"
return StreamingResponse(generate(), media_type="text/event-stream")
@app.delete("/session/{session_id}")
async def clear_session(session_id: str):
sessions.pop(session_id, None)
return {"status": "cleared"}Run it:
pip install fastapi uvicorn anthropic
uvicorn main:app --reloadSTEP 5: Cost Optimization
As conversations grow, token consumption compounds rapidly. Here are the key optimization techniques.
5-1: Summarize and Trim History
def summarize_and_trim(
client: anthropic.Anthropic,
history: list,
max_turns: int = 10
) -> list:
if len(history) <= max_turns * 2:
return history
old_history = history[:-max_turns * 2]
recent_history = history[-max_turns * 2:]
summary_response = client.messages.create(
model="claude-haiku-4-5-20251001", # Use cheap model for summarization
max_tokens=512,
messages=[{
"role": "user",
"content": f"Summarize this conversation in 3 sentences:\n\n{json.dumps(old_history)}"
}]
)
summary = summary_response.content[0].text
return [
{"role": "user", "content": f"[Conversation summary] {summary}"},
{"role": "assistant", "content": "Understood. I have context from our previous conversation."},
*recent_history
]5-2: Prompt Caching for Long System Prompts
If your system prompt is long (internal docs, specs), caching can cut costs by up to 90%.
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
system=[
{
"type": "text",
"text": "Your very long system prompt here (internal docs, product specs, etc.)...",
"cache_control": {"type": "ephemeral"}
}
],
messages=history
)5-3: Choose the Right Model for Each Task
| Use Case | Recommended Model | Reason |
|---|---|---|
| Simple Q&A | claude-haiku-4-5-20251001 | Fast, cheap |
| Coding assistance | claude-sonnet-4-6 | Best balance |
| Complex reasoning | claude-opus-4-6 | Highest accuracy |
| Summarization/classification | claude-haiku-4-5-20251001 | Cost optimal |
STEP 6: Error Handling and Retries — don't let everyone wait alike
This is the section I rewrote. My first pass was the straightforward exponential backoff below, where 2 ** attempt simply stretches the wait.
import anthropic
import time
from typing import Optional
client = anthropic.Anthropic(api_key="YOUR_API_KEY")
def safe_chat(
messages: list,
system: str = "",
max_retries: int = 3,
retry_delay: float = 1.0
) -> Optional[str]:
for attempt in range(max_retries):
try:
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=2048,
system=system,
messages=messages
)
return response.content[0].text
except anthropic.RateLimitError:
wait_time = retry_delay * (2 ** attempt)
print(f"Rate limited. Retrying in {wait_time}s...")
time.sleep(wait_time)
except anthropic.APIStatusError as e:
if e.status_code >= 500:
print(f"Server error ({e.status_code}). Retrying...")
time.sleep(retry_delay)
else:
print(f"Client error: {e.message}")
return None
except anthropic.APIConnectionError:
print("Connection error. Retrying...")
time.sleep(retry_delay)
print("Max retries reached")
return NoneIt worked, but failures arrived in clumps whenever traffic picked up. When I finally traced it, the problem wasn't in the code — it was in how everything waited. When every client waits by the same formula, every client comes back at the same instant. One second, two seconds, four seconds — marching in step, straight back into the same wall.
Measuring the two waiting strategies
I couldn't quite believe the difference, so I pulled the waiting logic out and counted it without touching the API. Fifty clients hit a 429 at the same moment, the server admits five requests per second, and each client gives up after five retries. The numbers below are averages over twenty runs with different random seeds.
| Waiting strategy | Peak arrivals in one second | Total API calls | Gave up | Median time to success |
|---|---|---|---|---|
2 ** attempt (no jitter) |
50.0 | 275.0 | 20.0 clients | 11.0s |
uniform(0, 2 ** attempt) (full jitter) |
8.3 | 107.5 | 0 | 0.6s |
The same number of retries, and the outcomes are nowhere near each other. Without jitter, all fifty return together, keep overflowing the five-per-second capacity, and twenty of them exhaust their attempts. Spreading the return times flattens the peak to about eight, cuts total calls to roughly forty percent, and nobody gives up. What the table told me was that scattering the return times mattered more than lengthening the wait.
The script is short, so I'd encourage you to swap in your own concurrency and capacity and see what falls out.
import random, statistics
N_CLIENTS, MAX_RETRIES, BASE, CAPACITY = 50, 5, 1.0, 5
def wait(strategy, attempt, rng):
if strategy == "plain":
return round(BASE * (2 ** attempt), 1) # everyone returns together
return round(rng.uniform(0, BASE * (2 ** attempt)), 1) # full jitter
def run(strategy, seed):
rng = random.Random(seed)
cs = [{"attempt": 0, "next_t": wait(strategy, 0, rng), "done": False}
for _ in range(N_CLIENTS)]
t, peak, calls, ok = 0.0, 0, N_CLIENTS, []
while t < 300 and any(not c["done"] for c in cs):
arrivals = [c for c in cs if not c["done"] and abs(c["next_t"] - t) < 1e-9]
rng.shuffle(arrivals)
peak = max(peak, len(arrivals))
accepted = 0
for c in arrivals:
calls += 1
if accepted < CAPACITY: # within capacity, request goes through
accepted += 1
c["done"] = True
ok.append(t)
else:
c["attempt"] += 1
if c["attempt"] > MAX_RETRIES:
c["done"] = True # gave up
else:
c["next_t"] = round(t + wait(strategy, c["attempt"], rng), 1)
t = round(t + 0.1, 1)
return peak, calls, len(ok), statistics.median(ok)
for s in ("plain", "full_jitter"):
r = [run(s, seed) for seed in range(2000, 2020)]
print(s, [round(statistics.mean(col), 1) for col in zip(*r)])Read the wait the server already told you
There's a second thing I rebuilt after the fact. The Claude API attaches a retry-after header to its 429 responses, and that number is more accurate than anything I can guess locally.
The part worth knowing is that not every 429 carries that header. A 429 raised by hitting a spend cap arrives without retry-after, and waiting won't change the outcome. The presence of the header is what separates "wait and it will pass" from "wait and it still won't" (Rate limits — Claude Platform Docs).
With both changes folded in, the retry logic settled into this shape:
import random
import time
from typing import Optional
import anthropic
client = anthropic.Anthropic(api_key="YOUR_API_KEY")
def _retry_after_seconds(err) -> Optional[float]:
# Return the wait the server asked for, if it named one
headers = getattr(getattr(err, "response", None), "headers", None) or {}
raw = headers.get("retry-after")
if raw is None:
return None
try:
return float(raw)
except (TypeError, ValueError):
return None
def safe_chat(
messages: list,
system: str = "",
max_retries: int = 5,
base_delay: float = 1.0,
max_delay: float = 30.0,
) -> Optional[str]:
for attempt in range(max_retries):
try:
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=2048,
system=system,
messages=messages,
)
return response.content[0].text
except anthropic.RateLimitError as e:
hinted = _retry_after_seconds(e)
if hinted is None and attempt >= 1:
# A spend-cap 429 won't clear by waiting, so hand it back early
print("429 without retry-after. Check your usage limits.")
return None
ceiling = min(hinted or base_delay * (2 ** attempt), max_delay)
wait_time = random.uniform(0, ceiling) # full jitter
print(f"Rate limited. Retrying in {wait_time:.1f}s")
time.sleep(wait_time)
except anthropic.APIStatusError as e:
if e.status_code >= 500:
wait_time = random.uniform(0, min(base_delay * (2 ** attempt), max_delay))
print(f"Server error ({e.status_code}). Retrying in {wait_time:.1f}s")
time.sleep(wait_time)
else:
# 4xx returns the same result however many times you send it
print(f"Client error: {e.message}")
return None
except anthropic.APIConnectionError:
wait_time = random.uniform(0, min(base_delay * (2 ** attempt), max_delay))
print(f"Connection error. Retrying in {wait_time:.1f}s")
time.sleep(wait_time)
print("Max retries reached")
return NoneThree changes, nothing more: honour the server's number, multiply the wait by a random factor, and tell apart the 429 that waiting won't fix. A retry protects the queue everyone shares in the next second, not just your own request. Since I started thinking about it that way, the shape of failures during busy hours has changed.
STEP 7: The Complete Production Class
Everything unified into a battle-tested class.
import anthropic
import uuid
from typing import Generator
class ProductionChatBot:
def __init__(
self,
api_key: str,
system_prompt: str = "You are a helpful assistant.",
model: str = "claude-sonnet-4-6",
max_tokens: int = 2048,
max_history_turns: int = 20,
):
self.client = anthropic.Anthropic(api_key=api_key)
self.system_prompt = system_prompt
self.model = model
self.max_tokens = max_tokens
self.max_history_turns = max_history_turns
self.history = []
self.session_id = str(uuid.uuid4())
self.total_tokens_used = 0
def send(self, message: str) -> str:
self.history.append({"role": "user", "content": message})
response = self.client.messages.create(
model=self.model,
max_tokens=self.max_tokens,
system=self.system_prompt,
messages=self.history
)
assistant_text = response.content[0].text
self.history.append({"role": "assistant", "content": assistant_text})
self.total_tokens_used += response.usage.input_tokens + response.usage.output_tokens
if len(self.history) > self.max_history_turns * 2:
self.history = self.history[-(self.max_history_turns * 2):]
return assistant_text
def stream(self, message: str) -> Generator[str, None, None]:
self.history.append({"role": "user", "content": message})
full_response = ""
with self.client.messages.stream(
model=self.model,
max_tokens=self.max_tokens,
system=self.system_prompt,
messages=self.history
) as stream:
for text in stream.text_stream:
full_response += text
yield text
self.history.append({"role": "assistant", "content": full_response})
def get_stats(self) -> dict:
return {
"session_id": self.session_id,
"turns": len(self.history) // 2,
"total_tokens": self.total_tokens_used,
"estimated_cost_usd": self.total_tokens_used / 1_000_000 * 3.0
}
def reset(self):
self.history = []
self.total_tokens_used = 0
# Example usage
bot = ProductionChatBot(
api_key="YOUR_API_KEY",
system_prompt="""You are a senior Python engineer with deep expertise
in clean code, performance optimization, and production best practices."""
)
print("Claude: ", end="")
for chunk in bot.stream("What are the best practices for building a FastAPI CRUD app?"):
print(chunk, end="", flush=True)
print()
print(f"\n📊 Stats: {bot.get_stats()}")STEP 8: Connecting to LINE, Telegram, and Discord — Platform-Specific Differences
So far we've run the chatbot from a terminal and our own web API. To put it in front of real users, you'll often want it on a messaging platform like LINE or Telegram.
When I first put a chatbot built as an indie developer onto LINE, the thing that caught me off guard was the reply token's short lifetime.
Both share the same skeleton: the platform POSTs to your server via a webhook, you verify the signature, you call Claude, and you reply. The differences live in the details — and that's where people get stuck.
LINE: the reply token has a short life
The first thing that trips people up with the LINE Messaging API is the reply token. A reply token can be used only once, and it expires quickly (about a minute after it's issued). A few seconds of Claude latency is fine, but if you add long generations or tool calls, the token can expire before you reply.
A stable pattern is a two-step approach: show a loading animation first to preserve the reply token, then send the answer with a push message once it's ready.
Signature verification is mandatory. If you don't verify the X-Line-Signature header, anyone can send forged events to your endpoint.
import hmac
import hashlib
import base64
import os
import requests
from fastapi import FastAPI, Request, HTTPException
app = FastAPI()
LINE_CHANNEL_SECRET = os.environ["LINE_CHANNEL_SECRET"]
LINE_ACCESS_TOKEN = os.environ["LINE_CHANNEL_ACCESS_TOKEN"]
def verify_line_signature(body: bytes, signature: str) -> bool:
"""Verify X-Line-Signature with HMAC-SHA256."""
digest = hmac.new(
LINE_CHANNEL_SECRET.encode("utf-8"), body, hashlib.sha256
).digest()
expected = base64.b64encode(digest).decode("utf-8")
return hmac.compare_digest(expected, signature)
def line_push(user_id: str, text: str) -> None:
"""Send via push message, which doesn't depend on a reply token."""
requests.post(
"https://api.line.me/v2/bot/message/push",
headers={"Authorization": f"Bearer {LINE_ACCESS_TOKEN}"},
json={"to": user_id, "messages": [{"type": "text", "text": text}]},
timeout=10,
)
@app.post("/line/webhook")
async def line_webhook(request: Request):
body = await request.body()
signature = request.headers.get("X-Line-Signature", "")
if not verify_line_signature(body, signature):
raise HTTPException(status_code=403, detail="signature mismatch")
events = (await request.json()).get("events", [])
for event in events:
if event.get("type") == "message" and event["message"]["type"] == "text":
user_id = event["source"]["userId"]
user_text = event["message"]["text"]
# Keep history per user (reuse ProductionChatBot from STEP 7)
reply = get_bot_for(user_id).send(user_text)
line_push(user_id, reply)
return {"ok": True}get_bot_for(user_id) is meant to look up a ProductionChatBot instance keyed by user ID. An in-process dictionary is wiped on restart, so in production it's safer to move conversation history to an external store like Redis.
Telegram: choose webhook or long polling
Telegram offers two ways to receive messages: long polling, where you loop on getUpdates, and webhook, where the platform POSTs to a public endpoint.
Polling is far easier during development — no public URL, no HTTPS certificate, it just runs locally. For a server that stays up in production, a webhook is leaner. When you use a webhook, register it with a secret_token and verify the X-Telegram-Bot-Api-Secret-Token header on incoming requests.
Since Claude takes a moment to respond, calling sendChatAction to show "typing…" makes the experience feel noticeably better.
import os
import requests
TG_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
API = f"https://api.telegram.org/bot{TG_TOKEN}"
def tg_typing(chat_id: int) -> None:
requests.post(f"{API}/sendChatAction",
json={"chat_id": chat_id, "action": "typing"}, timeout=10)
def tg_send(chat_id: int, text: str) -> None:
requests.post(f"{API}/sendMessage",
json={"chat_id": chat_id, "text": text}, timeout=10)
def run_polling():
"""Development: receive via long polling."""
offset = None
while True:
resp = requests.get(f"{API}/getUpdates",
params={"timeout": 30, "offset": offset}, timeout=40)
for update in resp.json().get("result", []):
offset = update["update_id"] + 1
msg = update.get("message")
if not msg or "text" not in msg:
continue
chat_id = msg["chat"]["id"]
tg_typing(chat_id)
reply = get_bot_for(str(chat_id)).send(msg["text"])
tg_send(chat_id, reply)Discord: You Have Three Seconds to Say "Got It"
LINE and Telegram both arrive over webhooks. I started writing the Discord integration with the same assumptions, and every slash command I built came back with nothing at all.
With discord.py, events don't arrive over a webhook. They come through a persistent WebSocket connection (the Gateway), so there's no signature to verify. Discord swaps that requirement for a different one: a slash command must acknowledge within three seconds, or the interaction is dead.
Claude takes a few seconds to generate. Write the obvious version — receive the question, call Claude, reply — and you will blow through that window almost every time.
The fix is ctx.defer(). Call it first and Discord shows a thinking state, which frees you to send the real answer later via followup.send(). It's close cousin to holding back the LINE reply token and switching to a push message.
The second constraint is the 2,000-character limit per message. Claude will sail past that if you let it, so nudge it in the system prompt and truncate on the way out anyway.
import discord
from discord import option
from discord.ext import commands
SYSTEM_PROMPT = """You are an AI assistant living in a Discord server.
Keep answers concise and mind the 2,000-character limit per message.
Always wrap code in code blocks."""
class ClaudeCog(commands.Cog):
def __init__(self, bot: commands.Bot):
self.bot = bot
@discord.slash_command(name="ask", description="Ask Claude a question")
@option("question", description="Your question")
async def ask(self, ctx: discord.ApplicationContext, question: str):
# Dodge the three-second deadline: acknowledge first.
await ctx.defer()
# Reuse the ProductionChatBot from STEP 7, keyed per user.
reply = get_bot_for(str(ctx.author.id)).send(question)
# Stay under the 2,000-character ceiling.
if len(reply) > 1900:
reply = reply[:1900] + "\n\n…(truncated for length)"
await ctx.followup.send(reply)
def setup(bot: commands.Bot):
bot.add_cog(ClaudeCog(bot))On the startup side, enable the Message Content Intent explicitly. Your code and the Developer Portal both have to agree, or the bot throws on boot and never comes up.
intents = discord.Intents.default()
intents.message_content = True # Turn this on in the Developer Portal too.
bot = commands.Bot(command_prefix="!", intents=intents)Three Discord-specific snags worth writing down, all of which I hit myself:
| Symptom | Cause | Fix |
|---|---|---|
PrivilegedIntentsRequired on startup |
Message Content Intent is off in the Developer Portal | Bot tab → Privileged Gateway Intents → enable Message Content Intent |
| Slash commands never appear in the picker | Invite URL is missing the applications.commands scope |
Add the scope in the OAuth2 URL Generator and re-invite the bot |
| Command edits don't show up | Global commands can take up to an hour to propagate | Register guild commands during development for instant updates |
That last one is unglamorous, and it cost me more time than the other two combined. I was convinced the code was broken and rewrote it twice, when all I was actually doing was waiting for propagation. Register guild-scoped commands while you develop. It changes the rhythm of the whole feedback loop.
The Three Platforms, Side by Side
Lined up together, the differences collapse into three things: how messages arrive, what your deadline is, and how you're expected to reply.
| LINE | Telegram | Discord | |
|---|---|---|---|
| Transport | Webhook | Webhook or long polling | Gateway (persistent WebSocket) |
| Inbound verification | X-Line-Signature, HMAC-SHA256 |
secret_token header comparison |
None needed (Gateway is authenticated) |
| Reply deadline | Reply token expires in ~1 minute | Effectively none | 3 seconds for slash commands |
| How you beat the deadline | Show loading, then push message | — | ctx.defer(), then followup.send() |
| Typing indicator | Loading animation API | sendChatAction |
channel.typing() |
| Per-message character cap | 5,000 | 4,096 | 2,000 |
What clicked for me, staring at that table, is that all three are asking for the same thing in different accents: prove your server is alive, and prove it fast. Try to wait for Claude to finish before you say anything, and you'll meet the same wall wearing a different name.
Acknowledge first, deliver the body afterward. Build it that way from the start and porting to the next platform stops being a rewrite.
STEP 9: Don't Send Raw Personal Data to Claude — PII Masking
When you route internal data or customer messages through a chatbot, personal information (PII) like email addresses and phone numbers flows straight into the API. Not wanting raw values to land in your input logs or model requests is a common production requirement.
A practical approach is reversible masking: replace PII with placeholders before sending, then restore it after the response comes back. Claude only ever sees sanitized strings like {{EMAIL_1}}, while the user gets a natural reply with the real values restored.
import re
class PIIMasker:
"""Reversibly mask PII before sending and restore it after."""
PATTERNS = {
"EMAIL": re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+"),
"PHONE": re.compile(r"\b\d{3}[-.]?\d{3,4}[-.]?\d{4}\b"),
"CARD": re.compile(r"\b(?:\d[ -]?){13,16}\b"),
}
def mask(self, text: str):
mapping = {}
counters = {}
def replace(kind, m):
counters[kind] = counters.get(kind, 0) + 1
token = f"{{{{{kind}_{counters[kind]}}}}}"
mapping[token] = m.group(0)
return token
for kind, pattern in self.PATTERNS.items():
text = pattern.sub(lambda m, k=kind: replace(k, m), text)
return text, mapping
def unmask(self, text: str, mapping: dict) -> str:
for token, original in mapping.items():
text = text.replace(token, original)
return text
# Usage
masker = PIIMasker()
user_text = "Send the invoice to tanaka@example.com and also call 555-123-4567"
masked, mapping = masker.mask(user_text)
print(masked)
# -> Send the invoice to {{EMAIL_1}} and also call {{PHONE_1}}
reply = bot.send(masked) # Claude only sees the masked text
restored = masker.unmask(reply, mapping) # the user gets the real values backOne honest caveat: regex-based masking is not a silver bullet. Fixed-format values like emails and phone numbers are easy to catch, but context-dependent PII such as names and addresses will slip through unless you add named-entity recognition (NER) or a dictionary.
If there's information that absolutely must not leave your systems, the most reliable design is to never send that field to Claude in the first place. Treat masking as one layer that reduces accidental leakage, and separate the data you truly need to protect at the input stage. With that two-layer mindset, you'll have a foundation you can explain even when an audit comes around.
One rung is enough for today
Nine steps is a lot to take in, and none of it has to be built at once. My own first version shipped at STEP 2. STEP 5 came after conversation history started costing real money, and STEP 6 came after the failures started arriving in clumps. Had I built them in a different order, I suspect I'd have tripped over the same two places anyway.
If you only touch one thing today, make it the random.uniform line in the STEP 6 retry. It drops into an existing bot in a couple of minutes, and it's the change whose effect I noticed first during busy hours.
- Running Claude API Parallel Tool Use in Production — giving your bot search and compute, and running several tools at once
- Claude Vision API in Production — when you want the bot to accept images and PDFs
- How I Cut My Claude API Bill in Half With Prompt Caching — when conversations grow and the bill starts to show
I haven't been tempted back to an off-the-shelf bot since that night. When you can see the inside, the one line you want to change is always within reach. Thank you for staying with a long article.