Application Routes

loglife.core.routes.webhook.routes

Webhook endpoint for inbound WhatsApp messages.

Receives POST requests, validates payloads, and enqueues messages for processing.

webhook()

Handle inbound WhatsApp messages.

Source code in src/loglife/core/routes/webhook/routes.py

@webhook_bp.route("/webhook", methods=["POST"])
def webhook() -> ResponseReturnValue:
    """Handle inbound WhatsApp messages."""
    try:
        data: dict = request.get_json()

        message = Message.from_payload(data)
        g.client_type = message.client_type  # expose client type to sender service

        enqueue_inbound_message(message)

        logger.info("Queued message type %s for %s", message.msg_type, message.sender)
        # For emulator, we don't need an explicit "queued" response message in the UI
        # We return an empty message so the emulator doesn't show "Message queued"
        return success_response(message="")
    except Exception as e:
        error = f"Error processing webhook > {e}"
        logger.exception(error)
        return error_response(error)

whatsapp_incoming()

Handle incoming messages from Meta WhatsApp Cloud API.

GET: Webhook verification (Meta sends verification challenge) POST: Processes text and interactive list messages, creates a custom payload, and forwards to /webhook. List selections are sent as text commands.

Source code in src/loglife/core/routes/webhook/routes.py

@webhook_bp.route("/whatsapp-incoming", methods=["GET", "POST"])
def whatsapp_incoming() -> ResponseReturnValue:
    """Handle incoming messages from Meta WhatsApp Cloud API.

    GET: Webhook verification (Meta sends verification challenge)
    POST: Processes text and interactive list messages, creates a custom payload,
          and forwards to /webhook. List selections are sent as text commands.
    """
    if request.method == "GET":
        return _handle_webhook_verification()

    # POST method - handle incoming messages
    try:
        data: dict = request.get_json()
        return _process_meta_message(data)
    except Exception:
        logger.exception("Error processing Meta webhook")
        return error_response("Error processing Meta webhook")

loglife.core.routes.emulator.routes

Web-based emulator for testing chat flows.

Serves the emulator UI and provides an SSE stream for realtime logs.

emulator()

Render the emulator HTML interface.

Source code in src/loglife/core/routes/emulator/routes.py

@emulator_bp.route("/")
def emulator() -> str:
    """Render the emulator HTML interface."""
    return render_template("emulator.html", db_url=EMULATOR_SQLITE_WEB_URL)

events()

Stream realtime log events to the browser via SSE.

Source code in src/loglife/core/routes/emulator/routes.py

@emulator_bp.route("/events")
def events() -> Response:
    """Stream realtime log events to the browser via SSE."""

    def stream() -> Generator[str, None, None]:
        # Listen yields messages from the broadcaster
        for msg in log_broadcaster.listen():
            # Handle multiline messages for SSE
            formatted_msg = msg.replace("\n", "\ndata: ")
            yield f"data: {formatted_msg}\n\n"

    response = Response(stream(), mimetype="text/event-stream")
    response.headers["Cache-Control"] = "no-cache"
    response.headers["X-Accel-Buffering"] = "no"
    return response

manage_prompts()

Read or update the prompts.json file.

Source code in src/loglife/core/routes/emulator/routes.py

@emulator_bp.route("/vapi-admin/api/prompts", methods=["GET", "POST"])
def manage_prompts() -> Response:
    """Read or update the prompts.json file."""
    try:
        prompts_path = APP_DIR / "config" / "prompts.json"

        if request.method == "GET":
            if not prompts_path.exists():
                return jsonify([]), 200

            with prompts_path.open(encoding="utf-8") as f:
                data = json.load(f)
            return jsonify(data)

        if request.method == "POST":
            new_prompts = request.get_json()
            if not isinstance(new_prompts, list):
                return jsonify({"error": "Prompts must be a list"}), 400

            # Validate structure (optional but recommended)
            # Expecting list of dicts with single key-value pair

            with prompts_path.open("w", encoding="utf-8") as f:
                json.dump(new_prompts, f, indent=2, ensure_ascii=False)

            return jsonify({"success": True, "prompts": new_prompts})

    except Exception as e:
        logger.exception("Error managing prompts")
        return jsonify({"error": str(e)}), 500

vapi_admin()

Render the VAPI assistant admin panel.

Source code in src/loglife/core/routes/emulator/routes.py

@emulator_bp.route("/vapi-admin")
def vapi_admin() -> str:
    """Render the VAPI assistant admin panel."""
    # Get assistant IDs from environment
    assistant_ids = {
        "1": os.getenv("NEXT_PUBLIC_VAPI_ASSISTANT_ID_1", ""),
        "2": os.getenv("NEXT_PUBLIC_VAPI_ASSISTANT_ID_2", ""),
        "3": os.getenv("NEXT_PUBLIC_VAPI_ASSISTANT_ID_3", ""),
        "4": os.getenv("NEXT_PUBLIC_VAPI_ASSISTANT_ID_4", ""),
    }
    return render_template("vapi-admin.html", assistant_ids=assistant_ids)

vapi_assistant(assistant_id)

Fetch or update assistant configuration from VAPI API.

Source code in src/loglife/core/routes/emulator/routes.py

@emulator_bp.route("/vapi-admin/api/assistant/<assistant_id>", methods=["GET", "PATCH"])
def vapi_assistant(assistant_id: str) -> Response:
    """Fetch or update assistant configuration from VAPI API."""
    vapi_private_key = os.getenv("VAPI_PRIVATE_KEY")

    if not vapi_private_key:
        return jsonify({"error": "VAPI_PRIVATE_KEY is not configured"}), 500

    if request.method == "GET":
        return _fetch_assistant(assistant_id, vapi_private_key)

    if request.method == "PATCH":
        return _update_assistant(assistant_id, vapi_private_key)

    return jsonify({"error": "Method not allowed"}), 405