Add CanLex Path B web app: Gradio thin client

.gitignore

@@ -0,0 +1,2 @@
+__pycache__/
+*.pyc

Dockerfile

@@ -0,0 +1,25 @@
+# CanLex Web (Path B) -- the private Gradio front-end.
+#
+# A thin client: it carries no corpus and loads no models. It calls the deployed
+# CanLex MCP server for retrieval and Google Gemini for answer composition.
+# Builds on Hugging Face Spaces (sdk: docker) or with plain Docker.
+FROM python:3.12-slim
+
+# Run as a non-root user (UID 1000) -- required by Hugging Face Spaces.
+RUN useradd --create-home --home-dir /app --uid 1000 app
+WORKDIR /app
+
+# Python dependencies first, so this layer caches across code changes.
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY --chown=app:app app.py .
+
+USER app
+ENV HOME=/app \
+    PORT=7860 \
+    PYTHONUNBUFFERED=1 \
+    GRADIO_ANALYTICS_ENABLED=False
+
+EXPOSE 7860
+CMD ["python", "app.py"]

README.md

@@ -1,11 +1,63 @@
 ---
-title: Canlex Web
-emoji: 🐠
-colorFrom: indigo
-colorTo: yellow
+title: CanLex Web
 sdk: docker
+app_port: 7860
 pinned: false
-short_description: Private web front-end for CanLex
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# CanLex Web (Path B)
+
+A private web front-end for **CanLex**, so people without an MCP-capable AI
+client can still ask Canadian legal-research questions.
+
+## How it works -- thin client
+
+This app holds **no copy of the legal corpus**. For each question it:
+
+1. calls the deployed **CanLex MCP server** (`canlex_search_legislation`) to
+   retrieve the cited source passages;
+2. sends those passages and the question to **Google Gemini Flash**, which
+   composes a grounded, cited answer following CanLex's own answering
+   instructions;
+3. displays the answer, with the retrieved passages shown for review.
+
+Because retrieval stays on the MCP server, a corpus or retrieval change is
+deployed once (to the MCP Space) and both the MCP connector and this website
+pick it up. Only UI or prompt changes redeploy this Space.
+
+## Required Space secrets
+
+Set these under **Settings -> Variables and secrets**:
+
+| Name | Kind | Purpose |
+|------|------|---------|
+| `GEMINI_API_KEY` | secret | Free Gemini key from Google AI Studio (https://aistudio.google.com/apikey). |
+| `CANLEX_WEB_AUTH` | secret | Login credentials, one `username:password` per line. |
+
+Optional overrides:
+
+| Name | Default |
+|------|---------|
+| `CANLEX_MCP_URL` | `https://beemer0-canlex.hf.space/mcp` |
+| `CANLEX_GEMINI_MODEL` | `gemini-2.5-flash` |
+
+If `CANLEX_WEB_AUTH` is unset the app falls back to an insecure default login
+(`canlex` / `canlex`) and logs a warning -- set the secret before real use.
+
+## Make the Space private
+
+Under **Settings -> Change Space visibility**, set the Space to **Private**.
+The app then has two layers of protection: Hugging Face gates who can open the
+page at all, and Gradio's username/password gates who can use it.
+
+## Run locally
+
+```
+pip install -r requirements.txt
+# PowerShell:
+$env:GEMINI_API_KEY = "your-key"
+$env:CANLEX_WEB_AUTH = "me:secret"
+python app.py
+```
+
+Then open http://localhost:7860.

app.py

@@ -0,0 +1,300 @@
+#!/usr/bin/env python3
+"""CanLex Web (Path B) -- a private web front-end for CanLex legal research.
+
+A thin client: it holds no copy of the legal corpus. For each question it
+  1. calls the CanLex MCP server (tool: canlex_search_legislation) to retrieve
+     the cited source passages;
+  2. sends those passages and the question to Google Gemini Flash, which
+     composes a grounded, cited answer following CanLex's answering instructions;
+  3. displays the answer, with the retrieved passages available for review.
+
+All configuration comes from environment variables, set as Hugging Face Space
+secrets. Run locally with:  python app.py
+"""
+import asyncio
+import json
+import os
+import sys
+import urllib.error
+import urllib.request
+from datetime import timedelta
+
+import gradio as gr
+from mcp import ClientSession
+from mcp.client.streamable_http import streamablehttp_client
+
+
+# --- Configuration (Hugging Face Space secrets / environment variables) -------
+
+# The deployed CanLex MCP server. Retrieval logic and the corpus live there; this
+# app never carries its own copy. Override only to point at a different server.
+MCP_URL = os.environ.get(
+    "CANLEX_MCP_URL", "https://beemer0-canlex.hf.space/mcp").strip()
+
+# Google Gemini -- the free-tier key is supplied as the GEMINI_API_KEY secret.
+GEMINI_MODEL = os.environ.get("CANLEX_GEMINI_MODEL", "gemini-2.5-flash").strip()
+GEMINI_ENDPOINT = ("https://generativelanguage.googleapis.com/v1beta/models/"
+                   f"{GEMINI_MODEL}:generateContent")
+
+SEARCH_TOOL = "canlex_search_legislation"
+DEFAULT_TOP_K = 6
+MAX_OUTPUT_TOKENS = 8192   # generous -- covers Gemini 2.5 thinking plus the answer
+REQUEST_TIMEOUT = 120      # seconds, applied to both the MCP and the Gemini calls
+
+
+def _load_auth() -> list[tuple[str, str]]:
+    """Parse CANLEX_WEB_AUTH (one 'username:password' per line) for Gradio auth."""
+    creds: list[tuple[str, str]] = []
+    for line in os.environ.get("CANLEX_WEB_AUTH", "").splitlines():
+        line = line.strip()
+        if not line or ":" not in line:
+            continue
+        user, password = (p.strip() for p in line.split(":", 1))
+        if user and password:
+            creds.append((user, password))
+    if not creds:
+        print("WARNING: CANLEX_WEB_AUTH is not set; using the insecure default "
+              "login 'canlex' / 'canlex'. Set CANLEX_WEB_AUTH as a Space secret "
+              "(one 'username:password' per line) before sharing this app.",
+              file=sys.stderr)
+        creds = [("canlex", "canlex")]
+    return creds
+
+
+AUTH = _load_auth()
+
+
+# --- Prompting ----------------------------------------------------------------
+
+# Light web framing only. The substantive answering guidance is CanLex's own
+# "ANSWERING INSTRUCTIONS", returned inline in the retrieved material below; this
+# system instruction just sets the web role and disables the tool-calling steps
+# that guidance assumes (this front-end cannot call follow-up MCP tools).
+SYSTEM_INSTRUCTION = """\
+You are CanLex Web, a Canadian legal-research assistant. A member of the public \
+has asked the legal question shown below through a web form. Compose one clear, \
+well-organised answer, grounded entirely in the retrieved source material that \
+follows the question.
+
+The retrieved material opens with a block headed "ANSWERING INSTRUCTIONS" from \
+the CanLex retrieval system. Follow those instructions, with these adjustments \
+for this web setting:
+
+- You have no tools and cannot retrieve anything further. Disregard any \
+instruction to call canlex_get_section, canlex_search_legislation or \
+canlex_case. Work only with the passages provided. If the question depends on a \
+provision, regulation or decision that is referred to but not reproduced below, \
+say so plainly and name what the reader should consult -- never guess its \
+contents.
+- Write for a reader who cannot see the raw passages: quote the key operative \
+words and give every citation in full.
+- Use plain Markdown -- short paragraphs, with headings or lists where they aid \
+clarity.
+- If the retrieved material does not actually answer the question, say so \
+directly rather than stretching it to fit.
+- Close with a one-line reminder that this is legal information, not legal \
+advice, and is current only to the dates stated in the sources."""
+
+
+# --- Retrieval: call the CanLex MCP server ------------------------------------
+
+async def _mcp_search(query: str, top_k: int) -> str:
+    """Call canlex_search_legislation on the remote MCP server; return its text."""
+    async with streamablehttp_client(
+        MCP_URL,
+        timeout=timedelta(seconds=REQUEST_TIMEOUT),
+        sse_read_timeout=timedelta(seconds=REQUEST_TIMEOUT),
+    ) as (read, write, _):
+        async with ClientSession(read, write) as session:
+            await session.initialize()
+            # The tool wraps its arguments in a single 'params' object (the
+            # server defines the tool as canlex_search_legislation(params: ...)).
+            result = await session.call_tool(
+                SEARCH_TOOL,
+                {"params": {"query": query, "top_k": int(top_k)}},
+            )
+    text = "\n".join(
+        block.text for block in result.content
+        if getattr(block, "type", None) == "text" and getattr(block, "text", None)
+    ).strip()
+    if result.isError:
+        raise RuntimeError(text or "the retrieval service returned an error.")
+    return text
+
+
+def retrieve(query: str, top_k: int) -> str:
+    """Synchronous wrapper around the async MCP retrieval call."""
+    return asyncio.run(_mcp_search(query, top_k))
+
+
+# --- Answer composition: call Google Gemini -----------------------------------
+
+def compose_answer(question: str, retrieved: str) -> str:
+    """Send the question and retrieved passages to Gemini; return the answer."""
+    api_key = os.environ.get("GEMINI_API_KEY", "").strip()
+    if not api_key:
+        raise RuntimeError(
+            "GEMINI_API_KEY is not set. Add it as a Space secret -- create a "
+            "free key at Google AI Studio (https://aistudio.google.com/apikey).")
+
+    user_prompt = (
+        "Answer this question for the user, using only the retrieved CanLex "
+        "material that follows it.\n\n"
+        f"QUESTION:\n{question}\n\n"
+        "RETRIEVED CANLEX MATERIAL (it begins with CanLex's own answering "
+        f"instructions):\n\n{retrieved}"
+    )
+    body = {
+        "systemInstruction": {"parts": [{"text": SYSTEM_INSTRUCTION}]},
+        "contents": [{"role": "user", "parts": [{"text": user_prompt}]}],
+        "generationConfig": {
+            "temperature": 0.2,
+            "maxOutputTokens": MAX_OUTPUT_TOKENS,
+        },
+        # Legal research routinely discusses crime, weapons and the like; relax
+        # the filters so legitimate legal text is not spuriously blocked.
+        "safetySettings": [
+            {"category": c, "threshold": "BLOCK_ONLY_HIGH"}
+            for c in ("HARM_CATEGORY_HARASSMENT", "HARM_CATEGORY_HATE_SPEECH",
+                      "HARM_CATEGORY_SEXUALLY_EXPLICIT",
+                      "HARM_CATEGORY_DANGEROUS_CONTENT")
+        ],
+    }
+    request = urllib.request.Request(
+        GEMINI_ENDPOINT,
+        data=json.dumps(body).encode("utf-8"),
+        headers={"Content-Type": "application/json", "x-goog-api-key": api_key},
+        method="POST",
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=REQUEST_TIMEOUT) as resp:
+            payload = json.loads(resp.read().decode("utf-8"))
+    except urllib.error.HTTPError as exc:
+        detail = exc.read().decode("utf-8", "replace")[:600]
+        raise RuntimeError(f"Gemini API returned HTTP {exc.code}: {detail}") \
+            from None
+    except urllib.error.URLError as exc:
+        raise RuntimeError(f"Could not reach the Gemini API: {exc.reason}") \
+            from None
+
+    candidates = payload.get("candidates") or []
+    if not candidates:
+        raise RuntimeError(
+            "Gemini returned no answer. promptFeedback: "
+            f"{payload.get('promptFeedback', {})}")
+    candidate = candidates[0]
+    parts = candidate.get("content", {}).get("parts", []) or []
+    answer = "".join(
+        part["text"] for part in parts
+        if "text" in part and not part.get("thought")
+    ).strip()
+    if not answer:
+        reason = candidate.get("finishReason", "unknown")
+        raise RuntimeError(
+            f"Gemini produced an empty answer (finishReason: {reason}). "
+            "If this is MAX_TOKENS, raise MAX_OUTPUT_TOKENS in app.py.")
+    return answer
+
+
+# --- Gradio handler -----------------------------------------------------------
+
+NO_RESULTS_PREFIX = "No results matched"
+ANSWER_PLACEHOLDER = "*Your answer will appear here.*"
+
+
+def answer(question: str, top_k: int):
+    """Generator: retrieve passages, compose an answer, stream status updates."""
+    question = (question or "").strip()
+    if not question:
+        yield "Please enter a legal question above.", ""
+        return
+
+    yield "_Retrieving source passages from CanLex..._", ""
+    try:
+        retrieved = retrieve(question, top_k)
+    except Exception as exc:
+        yield ("**Could not reach the CanLex retrieval service.**\n\n"
+               f"`{type(exc).__name__}: {exc}`\n\n"
+               "The service may be waking from sleep -- wait a moment and try "
+               "again."), ""
+        return
+
+    if not retrieved:
+        yield "The retrieval service returned nothing. Please try again.", ""
+        return
+    if retrieved.startswith(NO_RESULTS_PREFIX):
+        yield f"**No matching CanLex material was found.**\n\n{retrieved}", ""
+        return
+
+    yield "_Composing a grounded answer with Gemini..._", retrieved
+    try:
+        composed = compose_answer(question, retrieved)
+    except Exception as exc:
+        yield (f"**The answer could not be composed.**\n\n"
+               f"`{type(exc).__name__}: {exc}`"), retrieved
+        return
+
+    yield composed, retrieved
+
+
+# --- UI -----------------------------------------------------------------------
+
+INTRO = """\
+# CanLex -- Canadian Legal Research
+
+Ask a question about Canadian **border, customs, immigration, criminal, drug,
+labour or related federal law**. CanLex finds the governing statutory
+provisions, CBSA guidance, collective-agreement terms and leading court
+decisions, then composes an answer that cites them.
+
+Legal information, not legal advice -- always verify against the primary sources.
+"""
+
+EXAMPLE_QUESTIONS = [
+    "What are the detention review timelines for a permanent resident?",
+    "When is a foreign national inadmissible for serious criminality?",
+    "What overtime provisions apply to FB-group Border Services officers?",
+    "Can the CBSA seize goods for an undervalued customs declaration?",
+]
+
+with gr.Blocks(title="CanLex", analytics_enabled=False) as demo:
+    gr.Markdown(INTRO)
+
+    question = gr.Textbox(
+        label="Your legal question",
+        placeholder="e.g. What are the detention review timelines for a "
+                    "permanent resident?",
+        lines=3,
+    )
+    with gr.Accordion("Search options", open=False):
+        top_k = gr.Slider(
+            minimum=3, maximum=12, value=DEFAULT_TOP_K, step=1,
+            label="Number of source passages to retrieve",
+        )
+    with gr.Row():
+        submit = gr.Button("Ask CanLex", variant="primary")
+        clear = gr.Button("Clear")
+
+    gr.Examples(examples=EXAMPLE_QUESTIONS, inputs=question, label="Examples")
+
+    answer_md = gr.Markdown(value=ANSWER_PLACEHOLDER)
+    with gr.Accordion("Retrieved source passages", open=False):
+        sources_md = gr.Markdown()
+
+    submit.click(answer, [question, top_k], [answer_md, sources_md])
+    question.submit(answer, [question, top_k], [answer_md, sources_md])
+    clear.click(lambda: ("", ANSWER_PLACEHOLDER, ""), None,
+                [question, answer_md, sources_md])
+
+
+if __name__ == "__main__":
+    print(f"CanLex Web starting -- MCP: {MCP_URL}; model: {GEMINI_MODEL}; "
+          f"{len(AUTH)} login(s) configured.", file=sys.stderr)
+    demo.queue()
+    demo.launch(
+        server_name="0.0.0.0",
+        server_port=int(os.environ.get("PORT", "7860")),
+        auth=AUTH,
+        auth_message="Sign in to use CanLex.",
+        show_api=False,
+    )

requirements.txt

@@ -0,0 +1,5 @@
+# CanLex Web (Path B) -- thin-client dependencies.
+gradio>=5.0      # web UI with built-in username/password auth
+mcp>=1.9         # streamable-HTTP client for the CanLex MCP server
+# Google Gemini is called through its REST API using only the Python standard
+# library (urllib); no LLM SDK dependency is needed.