🏪 Sundry Shop Assistant — Source Code

def get_sales_by_category() -> dict: """Revenue and transaction count by product category, ranked best to worst.""" df = _load() grouped = df.groupby("Product Category")["Total Sales"].agg( ["sum", "count", "mean"] ) grouped = grouped.sort_values("sum", ascending=False) return { "categories": [ { "category": cat, "revenue_rm": round(float(row["sum"]), 2), "transactions": int(row["count"]), "avg_basket_rm": round(float(row["mean"]), 2), } for cat, row in grouped.iterrows() ] } # Public mapping — registered with Gemini via tool_bridge.py ALL_TOOLS = { "get_total_sales": get_total_sales, "get_top_day": get_top_day, "get_weekly_summary": get_weekly_summary, "get_sales_by_category": get_sales_by_category, "get_slowest_category": get_slowest_category, "compare_member_vs_visitor": compare_member_vs_visitor, "compare_gender": compare_gender, "get_payment_mix": get_payment_mix, "get_payment_by_customer_type": get_payment_by_customer_type, "get_basket_stats": get_basket_stats, }

from google.genai import types from mcp_tools import ALL_TOOLS _FUNCTION_DECLARATIONS = [ types.FunctionDeclaration( name="get_sales_by_category", description=( "Rank all product categories by revenue, including " "transaction count and average basket per category. " "Use this for 'kategori paling laku?' or 'top categories'." ), parameters=types.Schema(type=types.Type.OBJECT, properties={}), ), types.FunctionDeclaration( name="compare_member_vs_visitor", description=( "Compare Member vs Visitor customers: revenue share, " "average basket, and transaction count. Use this for " "loyalty program questions like 'member atau visitor " "spend lebih?'." ), parameters=types.Schema(type=types.Type.OBJECT, properties={}), ), # ... 8 more declarations ] def get_tools() -> list[types.Tool]: return [types.Tool(function_declarations=_FUNCTION_DECLARATIONS)] def get_tool_mapping() -> dict: return ALL_TOOLS

async for response in session.receive(): server_content = response.server_content tool_call = response.tool_call if server_content: if server_content.model_turn: for part in server_content.model_turn.parts: if part.inline_data: # Audio bytes streamed back to browser await audio_output_callback(part.inline_data.data) # Input/output transcripts for the UI transcript panel if server_content.input_transcription: await event_queue.put({ "type": "user", "text": server_content.input_transcription.text }) # Tool call handling — this is where MCP meets Live API if tool_call: function_responses = [] for fc in tool_call.function_calls: func_name = fc.name args = fc.args or {} if func_name in self.tool_mapping: tool_func = self.tool_mapping[func_name] loop = asyncio.get_running_loop() result = await loop.run_in_executor( None, lambda: tool_func(**args) ) function_responses.append( types.FunctionResponse( name=func_name, id=fc.id, response={"result": result}, ) ) # Send all tool results back; model resumes speaking await session.send_tool_response( function_responses=function_responses )

@app.websocket("/ws") async def websocket_endpoint(websocket: WebSocket): mode = websocket.query_params.get("mode", "audio").lower() response_modality = "TEXT" if mode == "text" else "AUDIO" await websocket.accept() audio_input_queue = asyncio.Queue() text_input_queue = asyncio.Queue() async def audio_output_callback(data: bytes): await websocket.send_bytes(data) gemini = GeminiLive( api_key=GEMINI_API_KEY, model=MODEL, response_modality=response_modality, voice_name=VOICE_NAME, ) async def receive_from_client(): while True: message = await websocket.receive() if message.get("bytes") is not None: # Raw PCM 16kHz from the mic await audio_input_queue.put(message["bytes"]) elif message.get("text") is not None: # JSON {"text": "..."} from keyboard input payload = json.loads(message["text"]) await text_input_queue.put(payload["text"]) # Run both loops in parallel — Gemini session drives the events async for event in gemini.start_session( audio_input_queue=audio_input_queue, text_input_queue=text_input_queue, audio_output_callback=audio_output_callback, ): await websocket.send_json(event)

async startAudio(onAudioData) { await this.initializeAudio(); this.mediaStream = await navigator.mediaDevices.getUserMedia({ audio: true, }); const source = this.audioContext.createMediaStreamSource(this.mediaStream); this.audioWorkletNode = new AudioWorkletNode( this.audioContext, "pcm-processor" ); this.audioWorkletNode.port.onmessage = (event) => { if (this.isRecording) { const downsampled = this.downsampleBuffer( event.data, this.audioContext.sampleRate, 16000 // Gemini Live expects 16kHz ); const pcm16 = this.convertFloat32ToInt16(downsampled); onAudioData(pcm16); // Sends bytes over WebSocket } }; source.connect(this.audioWorkletNode); this.isRecording = true; }

You are Adam, an AI assistant inside Sundry Shop Assistant for Malaysian sundry shop (kedai runcit) owners. ## Language style — Bahasa Malaysia santai (casual) Speak Bahasa Malaysia santai, NOT formal bahasa baku. The owner is a 52-year-old uncle running his own shop, not a government officer at a meeting. Santai style rules: - Use "tak" not "tidak", "dah" not "sudah", "ni" not "ini" - Softeners: "lah", "kan", "eh", "je", "Pak", "Bos" - Code-switch to English for business terms ("transaksi", "basket", "average", "member", "card", "cash", "compare") - Address the owner as "Pak" or "Bos" consistently Examples of formal vs santai: Formal (WRONG): "Jualan hari ini adalah sebanyak RM 847.23 dengan 23 transaksi." Santai (RIGHT): "Hari ni dah RM 847, Pak. Dua puluh tiga transaksi setakat ni." ## Never - Invent numbers — always call an MCP tool before stating a figure - Speak in long monologues — keep every answer under 30 seconds - Use heavy English jargon (revenue, conversion, SKU) when BM has natural equivalents (jualan, untung, barang)