"""Native tool-calling layer: schemas + executors for actions the chat model can take. Replaces the legacy fenced signal blocks (ticket_creation_signal, etc.). """ import asyncio import json import logging import uuid from datetime import datetime, timedelta, timezone from zoneinfo import ZoneInfo, ZoneInfoNotFoundError import anyio from app.core.config import settings from app.core.llm_client import client as openai_client from app.core.prompts import TICKET_CONSOLIDATION_PROMPT from app.services.infra.database import ( create_ticket, get_ticket, get_ticket_by_thread, insert_feedback, search_vector_data, update_ticket, ) from app.services.content.documents import download_link from app.services.messaging.email import email_configured, is_valid_email, send_email from app.services.chat.tools_settings import get_email_settings, TICKET_CATEGORIES, get_tool_settings from app.services.infra.embedding import get_embeddings from app.services.integrations.firestore import set_intervention_status from app.services.integrations.google_calendar import ( cancel_event as gcal_cancel_event, check_availability as gcal_check_availability, create_event as gcal_create_event, get_google_calendar_connection, ) from app.services.integrations.teams import ( cancel_event as teams_cancel_event, check_availability as teams_check_availability, create_event as teams_create_event, get_teams_connection, ) from app.services.catalog.products import ( match_products, semantic_products, to_card, ) from app.services.infra.quotas import QuotaExceededError from app.services.infra.redis import get_redis_client, publish_event logger = logging.getLogger(__name__) # An email whose body is one of these carries no information for the recipient. REFUSAL_MARKERS = ( "don't have enough information", "do not have enough information", ) TOOL_SCHEMAS = [ { "type": "function", "function": { "name": "search_knowledge_base", "description": ( "Search this business's knowledge base for facts. You have NO built-in knowledge " "about this business — its products, pricing, services, features, policies, hours, " "contact details or support processes are ONLY available through this tool. " "You MUST call it before answering any question about the business. " "If the results do not answer the question, call it again with a differently worded " "or more specific query before concluding you cannot answer. " "Returns matching content chunks, each with its source URL. " "Not for a question asking which product/service to get, including a goal or " "problem phrased with no product named ('how can I improve my brand health') -- " "that is recommend_products, even without shopping words in the sentence." ), "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "What to search for. Use the user's own wording, or a rephrasing " "targeting the specific fact you need.", }, }, "required": ["query"], }, }, }, { "type": "function", "function": { "name": "create_or_update_ticket", "description": ( "Create a support ticket, or update the session's existing ticket with a new issue. " "Use when the user reports an issue/complaint/bug, or wants to book a demo/meeting " "or contact the team/sales. " "ASK THE USER WHAT THE ISSUE IS before raising a ticket. Do not assemble the " "problem description from context or assume what went wrong — if they have only " "said something vague like 'it is broken', ask them what exactly is happening. " "The content field must be their own account of the problem, not your guess. " "Alongside that, collect their email address and phone number (and their name if " "not already known); ask for what is missing together rather than one question per " "turn. Never reuse an email or phone number that came from search results or a " "document — only what the user gives you here. " "You may write the short heading yourself, and set priority from how they describe " "the urgency. " "STOP ASKING once you have (a) their description of the problem and (b) their email " "and phone — call this tool immediately in that same turn. Do NOT then ask for a " "website URL, error messages, screenshots, what changed, or any other extra detail: " "the support team will follow up for those. One or two questions total, never an " "interrogation. " "If the session already has a ticket, do NOT re-ask contact details; just ask what " "the new issue is. " "Any team notification email is also sent in the background -- tell the user " "their ticket has been logged/raised, never that the team has already been " "emailed or has already seen it." ), "parameters": { "type": "object", "properties": { "user_name": {"type": "string", "description": "User's name"}, "heading": {"type": "string", "description": "Short ticket heading"}, "content": {"type": "string", "description": "Detailed issue description"}, "priority": {"type": "string", "enum": ["Low", "Medium", "High"]}, "email": {"type": "string", "description": "User's email address"}, "contact_no": {"type": "string", "description": "User's phone number"}, "any_other_contact_medium": {"type": "string", "description": "Optional other contact info"}, "category": { "type": "string", "enum": list(TICKET_CATEGORIES), "description": ( "Which team should handle this: General enquiries, Sales, " "or Business. Pick from the conversation's content." ), }, }, "required": ["heading", "content", "priority", "category"], }, }, }, { "type": "function", "function": { "name": "submit_feedback", "description": ( "Store the user's feedback about this conversation or the service. You MUST call this " "tool whenever the user expresses satisfaction or dissatisfaction — whether or not you " "asked them for it. That includes: any rating ('5 out of 5', '4/5', 'I'd give it a 3'), " "any verdict on whether you helped ('that was helpful', 'this didn't help at all', " "'you've been useless'), and any praise or complaint about the service. " "Thanking the user in text WITHOUT calling this tool loses their feedback permanently — " "the reply text is not stored anywhere as feedback. Set `question` to what you asked, or " "to a short description of what they were reacting to if they volunteered it unprompted. " "Do NOT call this for a plain 'thanks'/'ok' that carries no opinion about the service." ), "parameters": { "type": "object", "properties": { "question": {"type": "string", "description": "The feedback question asked"}, "answer": {"type": "string", "description": "The user's feedback answer"}, }, "required": ["question", "answer"], }, }, }, { "type": "function", "function": { "name": "request_human_takeover", "description": ( "Notify a real human team member to join this conversation. This is the ONLY way a " "human is actually notified — saying so in text without calling this tool notifies no one. " "Call it immediately, in the same turn, whenever the user explicitly REQUESTS a human/real " "person/agent/someone from the team (e.g. 'can I speak to someone', 'I want to talk to a " "real human', 'this isn't helping, connect me with your team'), or when the request is too " "complex or unsafe for you to handle safely. Do NOT call this for a QUESTION merely asking " "about human handoff capability (e.g. 'can this be escalated to a person if needed?', 'do " "you support human takeover?') — answer questions like that from context instead. " "Keep helping the user until the human joins." ), "parameters": { "type": "object", "properties": { "reason": {"type": "string", "description": "Why human help is needed"}, }, "required": ["reason"], }, }, }, { "type": "function", "function": { "name": "send_email", "description": ( "Email information to the user. Use ONLY when the user explicitly asks for " "something to be emailed to them (e.g. 'email me this', 'can you send that to " "my inbox', 'mail me the pricing'). " "Before calling: you MUST have searched the knowledge base for the content, and " "you MUST have the user's email address and have confirmed it back to them in " "your previous message ('Shall I send it to x@y.com?'). Never guess or invent an " "address, and never email an address the user did not give in this conversation. " "The `body` must be written ONLY from knowledge-base search results and what was " "already said in this conversation — never invent facts, prices or policies to " "fill out the email. If the knowledge base does not contain the information, tell " "the user instead of sending a made-up email. " "NEVER offer to email, and never call this tool, when your answer is that you " "could not find the information — an email saying you do not know is worse than " "no email. Offer a ticket or a human instead. " "Sending happens in the background after this call returns -- tell the user " "you are sending it now ('I'm sending that over now'), never that it has already " "arrived." ), "parameters": { "type": "object", "properties": { "to_email": { "type": "string", "description": "The recipient address, exactly as the user gave it.", }, "subject": { "type": "string", "description": "Short, specific subject line, e.g. 'GalaxiQ BrandForge pricing'.", }, "body": { "type": "string", "description": ( "Plain text body written from knowledge-base content. Use short " "paragraphs; '- ' at line start for bullets. Include the source URLs " "where relevant. No greeting placeholders like [Name]." ), }, }, "required": ["to_email", "subject", "body"], }, }, }, { "type": "function", "function": { "name": "attach_resources", "description": ( "Attach the source URLs/images your answer relied on so the frontend can render them as " "clickable buttons or image cards. Every result returned by search_knowledge_base " "carries a 'source' and may carry an 'images' list. " "ONLY attach a source that is a real web address starting with http:// or https://. " "A source that is an uploaded file name such as 'Pricing Sheet.docx' or 'policy.pdf' " "is NOT a URL -- never pass it here, because it produces a card that leads nowhere. " "To give the visitor an uploaded file, use send_document instead. " "You MUST call this tool, in the same turn as your text answer, whenever you used a " "search result that has a source URL or an image to produce your answer — this is REQUIRED, not " "optional, any time the context you drew on has a source. Never write the raw URL or " "image link directly into your reply text instead of calling this tool; the link belongs " "in the tool call, and your reply text should read naturally without it. Only include " "resources that directly relate to the current answer, and keep a URL and its matching " "image in the same object. Skip this tool only when no search result you actually used " "has a source URL or image." ), "parameters": { "type": "object", "properties": { "resources": { "type": "array", "items": { "type": "object", "properties": { "url": {"type": ["string", "null"]}, "image": {"type": ["string", "null"]}, "label": {"type": "string"}, }, "required": ["label"], }, }, }, "required": ["resources"], }, }, }, { "type": "function", "function": { "name": "send_document", "description": ( "Give the visitor the actual uploaded file — the brochure, price list, spec sheet or " "policy PDF itself — as a download link, rather than only describing what is in it. " "Call this when the visitor asks for the document ('do you have a brochure?', 'send me " "the pricing PDF', 'can I get that as a file?'), or when you have just answered from a " "document and handing them the original is obviously useful. " "Search the knowledge base FIRST: every result carries a 'source', and when that source " "is an uploaded file name (something ending in .pdf, .docx and so on) that is exactly " "what you pass as file_name. Never invent a file name — only use one you have actually " "seen in a search result. If the tool reports the document was not found, tell the " "visitor the file is not available and do not make up a link." ), "parameters": { "type": "object", "properties": { "file_name": { "type": "string", "description": ("Exact file name as it appeared in the 'source' field of a " "search result, e.g. 'pricing-2026.pdf'"), }, "reason": { "type": "string", "description": "Briefly, what the visitor asked for", }, }, "required": ["file_name"], }, }, }, { "type": "function", "function": { "name": "recommend_products", "description": ( "Show the visitor actual products, plans or services as cards they can act " "on, with the business's own buttons. " "Call this when the visitor is asking WHICH thing to get rather than asking " "ABOUT something: 'what do you have', 'show me', 'do you sell X', 'which plan " "suits me', anything with a budget or a category. This also covers advice-phrased " "requests for a recommendation, not only literal shopping requests -- 'what shirt " "should I wear', 'what should I get for a beach trip', 'what do you suggest for " "sensitive skin' are all still asking which product to get, just phrased as a " "question for advice; call this tool for those too, not search_knowledge_base. " "The same applies to a goal or outcome phrased as a problem, with NO product " "named -- 'how can I improve my brand health', 'how do I get more ROI from " "marketing', 'how do I grow my audience' are the visitor asking which of this " "business's products/services solves that, exactly the same as if they had " "said 'what do you have for X' -- call this tool, not search_knowledge_base, " "even though the sentence itself contains no shopping words. " "Do NOT call it for a question about a product they already named -- 'what " "does BrandForge do', 'does it integrate with Shopify' -- that is " "search_knowledge_base, and answering in prose is correct there. " "The tool returns only products this business actually has. If it returns " "none, say so plainly; never name a product, quote a price, or describe a " "button that did not come back from this tool." ), "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "What the visitor is looking for, in their own words", }, "category": { "type": "string", "description": "Category to narrow to, only if the visitor named one", }, }, "required": ["query"], }, }, }, { "type": "function", "function": { "name": "check_availability", "description": ( "Check whether one specific time the user asked for is open, e.g. '9 AM " "tomorrow'. Use whenever the user asks to book a call/meeting/demo -- never " "invent or claim a time is open without calling this first. " "If that exact time is free, this confirms it -- go ahead and book it once " "you also have the visitor's email and name. " "If it is NOT free, this returns a small number of real nearby alternatives " "instead -- relay those to the user; do not invent alternatives of your own, " "and do not read out more than the tool gives you. " "This tool's result always includes a 'timezone' field -- ALWAYS state that " "zone explicitly when you relay any time from this result (e.g. '10:30 AM " "IST', not just '10:30 AM'), so the user is never left guessing which " "timezone a time is in. Never assume it matches the timezone the user asked " "in -- use the value this tool actually returned. " "Give the date and time exactly as a person would say them (a 1-12 hour, a " "minute, and AM/PM) -- do NOT convert to 24-hour time yourself; this tool " "does that conversion for you so you never have to. " "BEFORE your FIRST check_availability call in a booking conversation: if the " "user has not already told you what timezone or city they're in, ASK them " "directly (e.g. 'What timezone are you in, so I book the right time for " "you?') -- do not silently assume they mean the business's own timezone, and " "do not guess. Only proceed without asking if they say they don't know or " "don't care (in which case leave visitor_timezone out and it defaults to the " "business's own zone). " "Once you know the user's timezone (asked or stated), pass it as " "visitor_timezone (an IANA zone name, e.g. 'Asia/Kolkata', 'Europe/London', " "'America/Los_Angeles') on EVERY check_availability and book_meeting call for " "the rest of THIS conversation -- never drop it on a later call, never switch " "between passing it and not passing it for what is meant to be the same " "request. Two calls for the same nominal time (e.g. both called '9 AM') must " "resolve to the exact same instant, or the user sees contradictory answers. " "If this returns no alternatives or an error, tell the user you could not " "find an open time and offer a ticket/human contact instead of inventing one." ), "parameters": { "type": "object", "properties": { "date": { "type": "string", "description": "ISO 8601 date (YYYY-MM-DD) the user wants, resolved " "from their natural-language request yourself (e.g. " "'tomorrow' -> tomorrow's date).", }, "hour_12": { "type": "integer", "description": "The hour the user asked for, on a 1-12 clock, exactly " "as you would say it out loud (e.g. 9 for '9 AM'). " "Never convert this to 24-hour time.", }, "minute": { "type": "integer", "description": "The minute the user asked for (0-59). Use 0 if they " "only gave an hour, e.g. 'tomorrow morning' -> a " "reasonable hour of your choosing with minute 0.", }, "meridiem": { "type": "string", "enum": ["AM", "PM"], "description": "Whether the requested time is AM or PM, exactly as " "you would say it out loud.", }, "visitor_timezone": { "type": "string", "description": "IANA timezone name (e.g. 'Asia/Kolkata') for the " "user's own timezone, established by asking them if " "they haven't already said. Once set for this " "conversation, pass the SAME value on every later call " "-- omit only if the user said they don't know/care.", }, }, "required": ["date", "hour_12", "minute", "meridiem"], }, }, }, { "type": "function", "function": { "name": "book_meeting", "description": ( "Book a meeting at a specific time. Only call this after check_availability " "has shown this exact time as open and the user has picked it -- never call " "this on a time you have not already confirmed is free. " "This books immediately; there is no separate confirmation step after this " "call succeeds, so only call it once the user has clearly picked a time, not " "while they are still considering options. " "You must have the visitor's email and name before calling -- ask for " "whichever is missing first. " "Give the date and time exactly as a person would say them (a 1-12 hour, a " "minute, and AM/PM) -- do NOT convert to 24-hour time yourself; this tool " "does that conversion for you so you never have to. " "Pass the EXACT SAME visitor_timezone value you used on the check_availability " "call that confirmed this slot (asked or stated earlier in this conversation) " "-- do NOT convert the hour yourself, and do NOT drop or change the timezone " "between check_availability and book_meeting for what is meant to be the same " "slot. Leave it out only if you never established one for this conversation. " "On success this returns 'confirmed_start' and 'timezone' -- when you tell " "the user their meeting is booked, state the time WITH that exact timezone " "(e.g. 'booked for 10:30 AM IST'), not a bare time -- never assume it matches " "whatever zone the user originally asked in. " "If this returns an error because the slot was taken in the meantime (someone " "else booked it first), tell the user and offer to check availability again -- " "never silently pick a different time yourself." ), "parameters": { "type": "object", "properties": { "date": { "type": "string", "description": "ISO 8601 date (YYYY-MM-DD) of the slot the user " "picked, matching the date on that slot as shown by " "check_availability.", }, "hour_12": { "type": "integer", "description": "The hour of the slot on a 1-12 clock, exactly as you " "would say it out loud (e.g. 2 for '2:30 PM'). Never " "convert this to 24-hour time.", }, "minute": { "type": "integer", "description": "The minute of the slot (0-59), e.g. 30 for '2:30 PM'.", }, "meridiem": { "type": "string", "enum": ["AM", "PM"], "description": "Whether the slot is AM or PM, exactly as you would " "say it out loud.", }, "visitor_timezone": { "type": "string", "description": "IANA timezone name (e.g. 'Asia/Kolkata') ONLY if the " "user explicitly named a timezone or city for this " "slot -- the same one passed to check_availability. " "Omit this field entirely otherwise.", }, "visitor_email": { "type": "string", "description": "The visitor's email address, as they gave it.", }, "visitor_name": { "type": "string", "description": "The visitor's name, as they gave it.", }, }, "required": ["date", "hour_12", "minute", "meridiem", "visitor_email", "visitor_name"], }, }, }, { "type": "function", "function": { "name": "cancel_meeting", "description": ( "Cancel a previously booked meeting. Ask for the visitor's email, and if they " "don't recall the exact time, roughly when it was (e.g. 'around 2pm ' or " "'sometime tomorrow') -- the lookup searches by these, not by an exact ID, so " "the more specific they are the more reliably it finds the right one. " "If this reports it could not find exactly one matching meeting, tell the " "user and ask for more detail (a narrower time) rather than trying again with " "the same information." ), "parameters": { "type": "object", "properties": { "visitor_email": { "type": "string", "description": "The email address the meeting was booked under.", }, "approximate_time": { "type": "string", "description": "ISO 8601 datetime near when the meeting was, if the " "user recalls it. Omit if they don't know.", }, }, "required": ["visitor_email"], }, }, }, ] async def execute_search_knowledge_base(args, ctx): """Retrieval as a model-chosen action. Returns content, not a status.""" query = (args.get("query") or "").strip() if not query: return {"results": [], "count": 0, "note": "Empty query."} try: embedding = get_embeddings(query, tenant_id=ctx["tenant_id"], user_id=ctx.get("user_id")) hits = search_vector_data(ctx["tenant_id"], embedding, limit=5) except QuotaExceededError: raise except Exception as ex: logger.error(f"Tool search_knowledge_base failed: {ex}", exc_info=True) return {"results": [], "count": 0, "note": "Search failed."} results = [] for hit in hits or []: meta = hit.get("metadata") or {} images = meta.get("top_images") or [] image_urls = [] for img in images[:3]: if isinstance(img, dict) and img.get("url"): image_urls.append(img["url"]) elif isinstance(img, str): image_urls.append(img) results.append({ "content": hit.get("content", ""), "source": meta.get("source") or meta.get("source_url") or "Internal Knowledge Base", "images": image_urls, }) return { "results": results, "count": len(results), "note": ("No matches. Try a different wording before telling the user you cannot answer." if not results else None), } async def execute_create_or_update_ticket(args, ctx): """Create a new ticket, or consolidate into the session's existing one.""" tenant_id = ctx["tenant_id"] thread_id = ctx["thread_id"] user_name = ctx.get("user_name") user_id = ctx.get("user_id") try: existing = get_ticket_by_thread(tenant_id, thread_id) if existing: try: cons_prompt = TICKET_CONSOLIDATION_PROMPT.format( old_heading=existing["heading"], old_content=existing["content"], new_heading=args["heading"], new_content=args["content"], ) cons_response = openai_client.chat.completions.create( model=settings.LLM_MODEL, messages=[{"role": "user", "content": cons_prompt}], max_completion_tokens=800, ) raw = cons_response.choices[0].message.content.strip() raw = raw.replace("```json", "").replace("```", "").strip() cons = json.loads(raw) update_data = { "heading": cons.get("heading", args["heading"]), "content": cons.get("content", f"{existing['content']}\n\n---\n{args['content']}"), "priority": args["priority"], "category": args.get("category") or existing.get("category") or "General enquiries", } except Exception as cons_ex: logger.error(f"Consolidation LLM call failed: {cons_ex}") update_data = { "heading": args["heading"], "content": f"{existing['content']}\n\n---\n[Update]:\n{args['content']}", "priority": args["priority"], "category": args.get("category") or existing.get("category") or "General enquiries", } success = update_ticket(tenant_id, existing["ticket_id"], update_data) if success: await _publish_ticket_event(tenant_id, existing["ticket_id"], "ticket_updated", f"Ticket {existing['ticket_id']} has been consolidated with new issues") # Notification failures must never reclassify an already-committed # update as a failure. update_data lacks email/contact_no/user_name # (update_ticket never touches those columns), so re-fetch the full # row rather than trying to reconstruct it from what's in hand. try: full_ticket = get_ticket(tenant_id, existing["ticket_id"]) if full_ticket: _notify_ticket(tenant_id, full_ticket, updated=True) except Exception as ex: logger.error(f"Ticket update notification failed for {existing['ticket_id']}: {ex}", exc_info=True) return {"status": "updated", "ticket_id": existing["ticket_id"]} return {"status": "error", "ticket_id": existing["ticket_id"]} ticket_data = dict(args) ticket_data["ticket_id"] = f"TICK-{uuid.uuid4().hex[:8].upper()}" ticket_data["category"] = args.get("category") or "General enquiries" if user_name: ticket_data["user_name"] = user_name if user_id: ticket_data["user_id"] = user_id new_id = create_ticket(tenant_id, ticket_data, thread_id=thread_id) if new_id: await _publish_ticket_event(tenant_id, ticket_data["ticket_id"], "new_ticket_created", f"A new ticket has been created: {ticket_data['ticket_id']}") # Notification failures must never reclassify an already-committed # ticket as a failed creation. try: _notify_ticket(tenant_id, ticket_data) except Exception as ex: logger.error(f"Ticket notification failed for {ticket_data['ticket_id']}: {ex}", exc_info=True) return {"status": "created", "ticket_id": ticket_data["ticket_id"]} return {"status": "error", "ticket_id": None} except QuotaExceededError: raise except Exception as ex: logger.error(f"Tool create_or_update_ticket failed: {ex}", exc_info=True) return {"status": "error", "ticket_id": None} async def _publish_ticket_event(tenant_id, ticket_id, event, message): """Notify the dashboard of a ticket change. publish_event() itself is fire-and-forget (see its docstring) -- a slow or stale Redis connection can no longer make this await hang, so this stays a plain await rather than wrapping it in a second background task. """ try: full = get_ticket(tenant_id, ticket_id) if full: await publish_event(f"tenant_{tenant_id}:tickets", {"type": "system", "event": event, "message": message, "data": full}) except Exception as ws_err: logger.warning(f"Failed to publish {event} to redis: {ws_err}") async def execute_submit_feedback(args, ctx): """Store explicit user feedback for this thread.""" tenant_id = ctx["tenant_id"] thread_id = ctx["thread_id"] user_name = ctx.get("user_name") history = ctx.get("history", []) try: feedback_data = {"question": args["question"], "answer": args["answer"], "thread_id": thread_id, "user_name": user_name} last_interaction = {} if history: last_interaction = { "last_user_query": next((m["content"] for m in reversed(history) if m["role"] == "user"), None), "last_ai_response": next((m["content"] for m in reversed(history) if m["role"] == "assistant"), None), } feedback_data["metadata"] = {"is_explicit_feedback": True, "rated_conversation": last_interaction} insert_feedback(tenant_id, feedback_data) return {"status": "saved"} except Exception as ex: logger.error(f"Tool submit_feedback failed: {ex}", exc_info=True) return {"status": "error"} async def execute_request_human_takeover(args, ctx): """Flag the thread for human intervention and notify the dashboard.""" tenant_id = ctx["tenant_id"] thread_id = ctx["thread_id"] user_id = ctx.get("user_id") try: set_intervention_status(tenant_id, thread_id, True, user_id=user_id) try: await publish_event(f"tenant_{tenant_id}:conversation_list", { "type": "system", "event": "intervention_requested", "message": f"AI has requested human assistance for thread {thread_id}", "data": {"thread_id": thread_id, "reason": args.get("reason"), "user_id": user_id}, }) except Exception as ws_err: logger.warning(f"Failed to publish intervention request to redis: {ws_err}") return {"status": "flagged"} except Exception as ex: logger.error(f"Tool request_human_takeover failed: {ex}", exc_info=True) return {"status": "error"} async def execute_attach_resources(args, ctx): """Resources are collected by the chat loop; this only acknowledges the call.""" resources = args.get("resources") count = len(resources) if isinstance(resources, list) else 0 return {"status": "attached", "count": count} def _email_count_key(tenant_id, thread_id): return f"chat_emails:{tenant_id}:{thread_id}" async def _emails_sent_in_thread(tenant_id, thread_id) -> int: """How many emails this conversation has already sent. Only consulted when a cap is configured. Fails CLOSED on a Redis outage: if we cannot prove the conversation is under the limit, we refuse rather than risk turning the chat into a mail relay. """ try: client = get_redis_client() value = await client.get(_email_count_key(tenant_id, thread_id)) return int(value or 0) except Exception as ex: logger.error(f"Could not read email count for {thread_id}: {ex}") return settings.EMAIL_MAX_PER_THREAD async def _record_email_sent(tenant_id, thread_id): key = _email_count_key(tenant_id, thread_id) try: client = get_redis_client() count = await client.incr(key) if count == 1: await client.expire(key, 60 * 60 * 24 * 7) # a conversation's lifetime except Exception as ex: logger.error(f"Could not record sent email for {thread_id}: {ex}") # Tasks created here are not awaited by anything, so nothing keeps a # reference to them once this module's local variable goes out of scope -- # without this set, asyncio may garbage-collect a task mid-send. See # https://docs.python.org/3/library/asyncio-task.html#asyncio.create_task _background_email_tasks = set() def _fire_and_forget_email(to_email, subject, body, tenant_id, first_name=None, on_sent=None): """Send in the background; the caller never waits on this. on_sent, if given, is an async callable run only after a confirmed successful send -- used to keep execute_send_email's per-thread rate counter accurate without blocking the tool-call round trip on it. """ async def _send(): try: result = await anyio.to_thread.run_sync( lambda: send_email(to_email, subject, body, first_name=first_name, tenant_id=tenant_id)) if result.get("status") == "sent": if on_sent: await on_sent() else: logger.error(f"Background email to {to_email} failed: {result.get('detail')}") except Exception as ex: logger.error(f"Background email to {to_email} raised: {ex}", exc_info=True) task = asyncio.create_task(_send()) _background_email_tasks.add(task) task.add_done_callback(_background_email_tasks.discard) return task def _ticket_team_email_content(ticket_data, updated=False): ticket_id = ticket_data["ticket_id"] verb = "Updated" if updated else "New" subject = f"{verb} {ticket_data.get('category')} ticket: {ticket_data['heading']}" lines = [ f"Ticket ID: {ticket_id}", f"Category: {ticket_data.get('category')}", f"Priority: {ticket_data.get('priority')}", f"From: {ticket_data.get('user_name') or 'Unknown'}", ] if ticket_data.get("email"): lines.append(f"Email: {ticket_data['email']}") if ticket_data.get("contact_no"): lines.append(f"Phone: {ticket_data['contact_no']}") lines.append("") lines.append(ticket_data.get("content") or "") return subject, "\n".join(lines) def _ticket_customer_email_content(ticket_data, updated=False): ticket_id = ticket_data["ticket_id"] if updated: subject = f"Your request has been updated — {ticket_id}" body = ( f"We've added your latest message to your existing ticket.\n\n" f"Ticket ID: {ticket_id}\n" f"Summary: {ticket_data['heading']}\n\n" f"Our team will follow up with you shortly." ) else: subject = f"We've received your request — {ticket_id}" body = ( f"Thanks for reaching out. Your ticket has been logged.\n\n" f"Ticket ID: {ticket_id}\n" f"Summary: {ticket_data['heading']}\n\n" f"Our team will follow up with you shortly." ) return subject, body def _notify_ticket(tenant_id, ticket_data, updated=False): """Email the matching business team, and confirm receipt to the customer. Called on both ticket creation and ticket update -- see the module's calling code. `updated` only changes the email wording (new vs. updated); the routing logic is identical either way. A category with no configured business email sends nothing to the team; that is intentional, not a bug (see the design spec). """ if not get_email_settings(tenant_id).get("enabled", True): return if not email_configured(tenant_id): return category = ticket_data.get("category") or "General enquiries" # Malformed stored entries are skipped rather than raising -- this runs after # the ticket is already committed. team_emails = [entry.get("email") for entry in get_tool_settings(tenant_id).get("ticket_emails", []) if isinstance(entry, dict) and entry.get("category") == category and entry.get("email")] # No per-thread rate cap on ticket notifications today (unlike execute_send_email's # EMAIL_MAX_PER_THREAD) -- deliberately deferred pending a design decision on the # right cap for this path. Tracked as a known gap, not fixed here. if team_emails: subject, body = _ticket_team_email_content(ticket_data, updated=updated) for address in team_emails: _fire_and_forget_email(address, subject, body, tenant_id) customer_email = (ticket_data.get("email") or "").strip() if customer_email and is_valid_email(customer_email): subject, body = _ticket_customer_email_content(ticket_data, updated=updated) _fire_and_forget_email(customer_email, subject, body, tenant_id, first_name=ticket_data.get("user_name")) async def execute_send_email(args, ctx): """Email knowledge-base content to an address the user gave in this conversation. Rate-limited per conversation so a misbehaving model cannot turn the chat into a mail relay. """ to_email = (args.get("to_email") or "").strip() subject = args.get("subject") or "" body = args.get("body") or "" if not get_email_settings(ctx["tenant_id"]).get("enabled", True): logger.warning(f"send_email called while disabled for {ctx['tenant_id']}") return {"status": "error", "detail": "Emailing is turned off for this business. Tell the user you cannot " "send emails."} if not email_configured(ctx["tenant_id"]): return {"status": "error", "detail": "Email is not configured. Tell the user you cannot email right now."} if not is_valid_email(to_email): return {"status": "error", "detail": f"'{to_email}' is not a valid email address. Ask the user to confirm it."} if not body.strip(): return {"status": "error", "detail": "Refusing to send an empty email."} # Emailing "I don't have enough information" wastes the customer's attention # and makes the product look broken. Nothing useful was found, so say so in # chat instead of posting it to their inbox. if any(marker in body.lower() for marker in REFUSAL_MARKERS): logger.info(f"Refusing to email a no-answer response for thread {ctx.get('thread_id')}") return {"status": "error", "detail": "There is nothing useful to email — the knowledge base did not answer " "this. Tell the user you could not find the information, and offer to " "raise a ticket or connect them with the team instead."} # EMAIL_MAX_PER_THREAD <= 0 disables the cap entirely. # The counter must survive across turns: tool_ctx is rebuilt on every message, # so an in-memory count would only ever limit a single turn. limit = settings.EMAIL_MAX_PER_THREAD already_sent = await _emails_sent_in_thread(ctx["tenant_id"], ctx["thread_id"]) if limit > 0 else 0 if limit > 0 and already_sent >= limit: logger.warning( f"Email limit ({settings.EMAIL_MAX_PER_THREAD}) reached for thread {ctx.get('thread_id')}" ) return {"status": "error", "detail": "Email limit reached for this conversation. Tell the user you cannot " "send any more emails in this chat."} tenant_id = ctx["tenant_id"] thread_id = ctx["thread_id"] async def _on_sent(): logger.info(f"Chat emailed {to_email} for thread {thread_id}") # Count the attempt, not the confirmed success: the send completes seconds # later in the background, so incrementing there would let rapid or # concurrent calls all read a stale count and blow past the cap. await _record_email_sent(tenant_id, thread_id) _fire_and_forget_email(to_email, subject, body, tenant_id, first_name=ctx.get("user_name"), on_sent=_on_sent) return {"status": "queued", "to": to_email} async def execute_send_document(args, ctx): """Turn a file name into a temporary download link the visitor can click.""" file_name = (args.get("file_name") or "").strip() if not file_name: return {"status": "error", "detail": "No file name given."} link = await anyio.to_thread.run_sync( lambda: download_link(ctx["tenant_id"], file_name)) if not link: logger.info(f"send_document: no stored file matching {file_name!r} " f"for {ctx['tenant_id']}") return {"status": "not_found", "detail": f"No uploaded document named {file_name}. " "Tell the visitor the file is not available."} logger.info(f"send_document: {link['file_name']} for thread {ctx.get('thread_id')}") return {"status": "sent", **link} async def execute_recommend_products(args, ctx): """Return products from the tenant's own catalogue. Never invents one.""" query = (args.get("query") or "").strip() # Meaning, not words. The query is embedded and compared against the # products' own embeddings, which the build writes for every product # whatever its source -- so "a gift for a coffee lover" reaches coffee, a # cup and a mug stand, and a catalogue from a spreadsheet or a product API # is as findable as a crawled one. # # The literal match is the fallback rather than the primary: it is there # for the case semantics handles worst, an exact product name or SKU the # visitor has typed verbatim. rows = await anyio.to_thread.run_sync( lambda: semantic_products(ctx["tenant_id"], query, limit=3)) if not rows: rows = await anyio.to_thread.run_sync( lambda: match_products(ctx["tenant_id"], query, limit=3)) if not rows: logger.info(f"recommend_products: nothing matched {query!r} " f"for {ctx['tenant_id']}") return {"status": "none", "products": [], "detail": "This business has no matching products. Tell the visitor " "plainly and do not invent one."} products = [to_card(r) for r in rows] logger.info(f"recommend_products: {len(products)} for thread {ctx.get('thread_id')}") return {"status": "found", "products": products} # check_availability only ever returns real nearby alternatives to the exact # time the user asked for (never a whole day/week dump), so this cap just # bounds how many of those alternatives get serialized. _MAX_AVAILABILITY_SLOTS = 4 # How far around the user's requested time to look for a nearby alternative # when that exact time isn't free. _NEARBY_SEARCH_WINDOW = timedelta(hours=4) def _resolve_effective_timezone(args: dict, business_tz): """The zone a requested time should be interpreted in: the visitor's explicitly-named zone if the model passed one and it's a real IANA name, otherwise the business's own connected-calendar zone. Callers must relay this SAME zone back in their response to the model (as the tool result's own "timezone" field) rather than always reporting the business's zone -- live testing found the tool result silently labelling a visitor-timezone request with the business's own zone name, which the model then relayed as if it matched the times it just gave, when the times were actually in a different zone entirely. """ visitor_tz_name = (args.get("visitor_timezone") or "").strip() if visitor_tz_name: try: return ZoneInfo(visitor_tz_name) except (ZoneInfoNotFoundError, ValueError, KeyError): pass return business_tz def _resolve_requested_time(args: dict, business_tz): """The date/hour_12/minute/meridiem the model gave us, as a real datetime -- or None if any field is missing or out of range. The model gives us the time exactly as a person would say it (e.g. 9, 30, "AM") and we do the 12-hour-to-24-hour conversion ourselves, deterministically, rather than asking the model to compute and pass a 24-hour datetime. Live testing found the model unreliable at that specific conversion (a systematic AM/PM slip), even after the tool description demanded it copy a value verbatim. Built in whatever _resolve_effective_timezone resolves to -- the visitor's explicitly-named zone if given, otherwise the business's own. """ tz = _resolve_effective_timezone(args, business_tz) try: day = datetime.fromisoformat(args.get("date") or "").date() hour_12 = int(args.get("hour_12")) minute = int(args.get("minute")) meridiem = (args.get("meridiem") or "").strip().upper() if not (1 <= hour_12 <= 12) or not (0 <= minute <= 59) or meridiem not in ("AM", "PM"): return None hour_24 = hour_12 % 12 if meridiem == "PM": hour_24 += 12 return datetime(day.year, day.month, day.day, hour_24, minute, tzinfo=tz) except (TypeError, ValueError): return None def _book_meeting_connection(tenant_id: str): """The active meeting provider's connection, or (None, None, error_dict) if the feature is off, nothing is selected, or the selected provider has no active connection. Mirrors execute_send_email's guard shape.""" tool_settings = get_tool_settings(tenant_id) or {} features = tool_settings.get("features") or {} if not features.get("book_meeting"): return None, None, {"status": "error", "detail": "Meeting booking is turned off for this business."} providers = tool_settings.get("providers") or {} provider = providers.get("book_meeting") if provider not in ("google_calendar", "teams"): return None, None, {"status": "error", "detail": "No meeting provider is connected for this business."} connection = (get_google_calendar_connection(tenant_id) if provider == "google_calendar" else get_teams_connection(tenant_id)) if not connection: return None, None, {"status": "error", "detail": "The connected meeting provider is unavailable right now."} return provider, connection, None async def execute_check_availability(args, ctx): """Checks whether the exact time the user asked for is free on whichever meeting provider this tenant has active -- and if not, real nearby alternatives instead of a generic list of whatever's chronologically first in the day. An unrecognized timezone string degrades to UTC rather than failing the call. """ provider, connection, error = await anyio.to_thread.run_sync( lambda: _book_meeting_connection(ctx["tenant_id"])) if error: return error tz_name = connection.get("timezone") or "UTC" try: tz = ZoneInfo(tz_name) except (ZoneInfoNotFoundError, ValueError, KeyError): logger.warning(f"Unrecognized timezone {tz_name!r} for tenant " f"{ctx['tenant_id']}; using UTC for the search window.") tz = timezone.utc requested_start = _resolve_requested_time(args, tz) if requested_start is None: return {"status": "error", "detail": "date, hour_12, minute, and meridiem are required and must be valid."} requested_end = requested_start + timedelta(minutes=30) # The zone the request was actually built in -- the visitor's explicit # zone if they named one, otherwise the business's own. This is what # must be relayed back, NOT always connection's own zone, or a # visitor-timezone request gets labelled with the wrong zone name. effective_tz_name = getattr(requested_start.tzinfo, "key", tz_name) check_fn = gcal_check_availability if provider == "google_calendar" else teams_check_availability now = datetime.now(timezone.utc) if requested_start >= now: exact_slots = await anyio.to_thread.run_sync( lambda: check_fn(connection, requested_start, requested_end + timedelta(minutes=1)) or []) if any(isinstance(s, dict) and s.get("start") == requested_start for s in exact_slots): return {"status": "ok", "timezone": effective_tz_name, "available": True, "requested": {"start": requested_start.isoformat(), "end": requested_end.isoformat()}} # Not free (or already past) -- look for real alternatives near the # requested time rather than dumping whatever's chronologically first in # the day, which is how a "9 AM" request used to surface 12:30-4:30 AM. search_start = max(now, requested_start - _NEARBY_SEARCH_WINDOW) search_end = requested_start + _NEARBY_SEARCH_WINDOW nearby = await anyio.to_thread.run_sync( lambda: check_fn(connection, search_start, search_end) or []) nearby = [s for s in nearby if isinstance(s, dict) and s.get("start") and s["start"] >= now] nearby.sort(key=lambda s: abs((s["start"] - requested_start).total_seconds())) nearby = nearby[:_MAX_AVAILABILITY_SLOTS] return {"status": "ok", "timezone": effective_tz_name, "available": False, "nearby_slots": [{"start": s.get("start").isoformat() if s.get("start") else None, "end": s.get("end").isoformat() if s.get("end") else None} for s in nearby if isinstance(s, dict)]} async def execute_book_meeting(args, ctx): """Books a slot immediately, after re-confirming it is still free.""" visitor_email = (args.get("visitor_email") or "").strip() visitor_name = (args.get("visitor_name") or "").strip() if not visitor_email or not visitor_name: return {"status": "error", "detail": "visitor_email and visitor_name are both required."} provider, connection, error = await anyio.to_thread.run_sync( lambda: _book_meeting_connection(ctx["tenant_id"])) if error: return error tz_name = connection.get("timezone") or "UTC" try: tz = ZoneInfo(tz_name) except (ZoneInfoNotFoundError, ValueError, KeyError): tz = timezone.utc start = _resolve_requested_time(args, tz) if start is None: return {"status": "error", "detail": "date, hour_12, minute, and meridiem are required and must be valid."} effective_tz_name = getattr(start.tzinfo, "key", tz_name) if start < datetime.now(timezone.utc): return {"status": "error", "detail": "That time has already passed."} end = start + timedelta(minutes=30) check_fn = gcal_check_availability if provider == "google_calendar" else teams_check_availability free_slots = await anyio.to_thread.run_sync( lambda: check_fn(connection, start, end + timedelta(minutes=1)) or []) still_free = any(isinstance(s, dict) and s.get("start") == start for s in free_slots) if not still_free: return {"status": "error", "detail": "That time is no longer available. Offer to check availability again."} create_fn = gcal_create_event if provider == "google_calendar" else teams_create_event booking = await anyio.to_thread.run_sync( lambda: create_fn(connection, start, end, visitor_email, visitor_name)) if not booking: return {"status": "error", "detail": "Could not book that meeting right now."} return {"status": "ok", "event_id": booking.get("event_id"), "meeting_link": booking.get("meeting_link"), "timezone": effective_tz_name, "confirmed_start": start.isoformat()} async def execute_cancel_meeting(args, ctx): """Finds and cancels a previously booked meeting.""" provider, connection, error = await anyio.to_thread.run_sync( lambda: _book_meeting_connection(ctx["tenant_id"])) if error: return error near_time = None if args.get("approximate_time"): try: near_time = datetime.fromisoformat(args["approximate_time"]) except (TypeError, ValueError): near_time = None cancel_fn = gcal_cancel_event if provider == "google_calendar" else teams_cancel_event cancelled = await anyio.to_thread.run_sync( lambda: cancel_fn(connection, args.get("visitor_email", ""), near_time=near_time)) if not cancelled: return {"status": "error", "detail": "Could not find exactly one matching meeting to cancel."} return {"status": "ok"} TOOL_EXECUTORS = { "search_knowledge_base": execute_search_knowledge_base, "create_or_update_ticket": execute_create_or_update_ticket, "submit_feedback": execute_submit_feedback, "request_human_takeover": execute_request_human_takeover, "send_email": execute_send_email, "attach_resources": execute_attach_resources, "send_document": execute_send_document, "recommend_products": execute_recommend_products, "check_availability": execute_check_availability, "book_meeting": execute_book_meeting, "cancel_meeting": execute_cancel_meeting, } async def dispatch_tool(name, args, ctx): """Route a model tool call to its executor. Returns a JSON-serializable result dict.""" handler = TOOL_EXECUTORS.get(name) if handler is None: logger.error(f"Unknown tool requested by model: {name}") return {"status": "error", "detail": "unknown tool"} return await handler(args, ctx)