I built AI Agent Convo, a Python WhatsApp assistant for a fictional clinic, in the aiagentconvo repository on my @NitinSukthe-G GitHub. It answers questions from a small clinic knowledge file, manages appointments in MongoDB, supports human handoff, and replies in English, Hinglish, or Telugu according to how a person writes. I wrote and unit-tested all 6 build phases (40 tests), but I have not deployed it or run it against a live WhatsApp number.
WhatsApp sends POST /webhook to FastAPI in app/main.py. The webhook verifies X-Hub-Signature-256, parses the message, pushes it to Redis’s incoming_queue, and returns HTTP 200 quickly. A separate workers/queue.py process consumes messages with BRPOP; it handles deduplication and session state, runs the Gemini agent loop, invokes tools backed by MongoDB collections for doctors, slots, appointments, and conversations, and sends WhatsApp replies. I use a separate worker rather than a FastAPI BackgroundTask so slow or failure-prone work does not delay the webhook and the worker can scale independently.
- I implemented these assistant tools:
get_clinic_info(topic)for timings, location, doctors, services, parking, insurance, and payment methods;list_doctors()andlist_available_slots(doctor_id, date)for WhatsApp lists;get_my_appointments()for the current number’s upcoming bookings;cancel_appointment(appointment_id)to cancel and free a slot; andrequest_human_handoff(reason)to silence the bot and alert staff.
I intercept book_appointment(doctor_id, date, time, patient_name) rather than letting the model book directly. The agent saves the proposed booking on the session and the worker sends Confirm/Cancel buttons; only a tapped Confirm executes the booking. I chose this deterministic checkpoint because booking costs money and cannot be undone by re-asking. Buttons and lists also reduce parsing errors, clarify choices, and let a tap complete an action without another model round-trip.
- I use Redis
SETNX msg:{wamid}with a 24-hour TTL to deduplicate WhatsApp deliveries; the worker silently drops repeats. Outgoing WhatsApp sends inwhatsapp/client.pyget 3 attempts with 1s/2s/4s backoff. Booking claims a slot with one atomicfind_one_and_update({..., booked: False}, {"$set": {"booked": True}})rather than a read-then-write, preventing two concurrent confirmations from claiming the same slot. The tests includetest_worker.py::test_duplicate_message_is_processed_onceandtests/test_bookings.py::test_double_booking_the_same_slot_is_rejected.
I support handoff both from free text such as “talk to a human” or “agent” and from request_human_handoff. The Redis session’s handoff flag stops the bot from running the agent or replying while staff handle the conversation; incoming messages are still stored. Staff receive a WhatsApp alert at STAFF_PHONE with the reason and last 5 messages. POST /admin/release/{phone}, protected by X-Admin-Token, clears the flag and returns the conversation to the bot.
A small Gemini call classifies each text message as en, hinglish, or te, but does not classify button or list taps. I cache the classification for 10 messages. The system prompt asks for replies in the detected language and script: Hinglish remains in Roman script and Telugu uses Telugu script. I use the per-language agent/prompts.py::UI_TEXT dictionary for button and list labels instead of live translation, keeping those labels consistent without another model call.
I accumulate each Gemini response’s usage_metadata token counts per phone number in MongoDB’s conversations collection and calculate cost using services/costs.py::PRICE_PER_MILLION_TOKENS_USD. I need to verify that price table against Gemini’s current pricing before trusting its dollar figures. GET /admin/conversations/{phone}, protected by X-Admin-Token, returns the running total alongside history and bookings. After deployment, I still plan to add a real-conversation cost screenshot and a short WhatsApp booking-flow GIF.
cp .env.example .env
uv sync
uv run python scripts/seed.py
uv run pytest -q
uv run uvicorn app.main:app --reload
uv run python -m workers.queue
cloudflared tunnel --url http://localhost:8000
The seed script creates 3 doctors and 7 days of slots; uv run pytest -q runs the 40 tests without live credentials. My sandbox had neither passwordless sudo nor a Docker daemon, so I could not install Redis there; I can install it with sudo apt install -y redis-server or point REDIS_URL at reachable Redis. MongoDB can use a free Atlas cluster. To test WhatsApp locally, I can expose the webhook with cloudflared tunnel --url http://localhost:8000, register its printed https://*.trycloudflare.com/webhook URL in Meta’s Business app under WhatsApp product → Configuration → Webhook, subscribe to the messages field, and message the test number.
- For a live setup I need Meta’s
PHONE_NUMBER_ID,WABA_ID,ACCESS_TOKEN, andAPP_SECRET; a randomVERIFY_TOKENentered in both.envand Meta’s webhook config;GEMINI_API_KEYfrom Google AI Studio;MONGODB_URIfrom a MongoDB Atlas free-cluster connection string;STAFF_PHONEfor handoff alerts; and a randomADMIN_TOKENprotecting/admin/*. Deployment is not done, and the app has not yet been tested against a live WhatsApp number.
I plan to deploy using the standing production-box process in ~/.claude/NEW-SERVER-RUNBOOK.md, pushing to a private repository with git push origin main, rather than using the Railway/Render suggestion in the original spec. Before deployment I must fill the .env heredoc in .github/workflows/deploy.yml, add the server private key to its SSH_KEY block (runbook §6 Step 3 requires a human), verify port 8004 is free with ss -ltnp, add the DNS A record for aiagentconvo.powersmy.biz, push and monitor the workflow, and then point Meta’s webhook at the production URL. The workflow deploys the aiagentconvo FastAPI service behind Caddy and the separate aiagentconvo-worker queue consumer as two systemd units. Redis will run as a native apt package on the box; it currently hosts no stateful services, and this app’s Redis use is small enough that a managed service would be unnecessary overhead.