Search Questions

Instructions

Implementation Reference

• mcp_server.py:512-516 (registration)

  Tool registration for search_questions using the @mcp.tool decorator with title 'Search Questions', tags {'search', 'core'}, and readOnly/idempotent annotations.

  @mcp.tool(
      title="Search Questions",
      tags={"search", "core"},
      annotations={"readOnlyHint": True, "idempotentHint": True},
  )

• mcp_server.py:517-528 (schema)

  Schema/type signature for search_questions defining its parameters: query (required), subject, paper, year, session, chapter, mode (default hybrid), limit (default 20), offset (default 0), expand (default True). Returns ToolResult.

  def search_questions(
      query: str,
      subject: Optional[str] = DEFAULT_SUBJECT,
      paper: Optional[int] = None,
      year: Optional[int] = None,
      session: Optional[str] = None,
      chapter: Optional[int] = None,
      mode: str = "hybrid",
      limit: int = 20,
      offset: int = 0,
      expand: bool = True,
  ) -> ToolResult:

• mcp_server.py:517-632 (handler)

  Full handler implementation of search_questions. Validates mode, applies spell correction, calls the /search API endpoint with filters (subject, paper, year, session, chapter), processes results into card items with metadata, and returns a ToolResult containing both a human-readable summary and structured_content payload with recommended_ids for follow-up get_questions calls.

  def search_questions(
      query: str,
      subject: Optional[str] = DEFAULT_SUBJECT,
      paper: Optional[int] = None,
      year: Optional[int] = None,
      session: Optional[str] = None,
      chapter: Optional[int] = None,
      mode: str = "hybrid",
      limit: int = 20,
      offset: int = 0,
      expand: bool = True,
  ) -> ToolResult:
      """Search past-paper questions with optional filters.
  
      Returns ranked questions with compact metadata. Each result includes:
      - Paper identification (paper, year, session, variant, paper_label)
      - Question number, marks, relevance score
      - Whether it's image-based (if so, question_image_url and ms_image_url are provided)
      - topic_signal: exam frequency and importance summary
      - question_url: link to full question page
  
      NEXT STEP: Call get_questions(question_ids_list=[...]) with the recommended_ids
      to get full question text, mark scheme key points, and images.
      """
      if not _validate_mode(mode):
          raise ToolError(
              "INVALID_MODE: mode must be one of 'hybrid', 'keyword', or 'semantic'."
          )
  
      capped_limit = max(1, min(limit, MAX_SEARCH_LIMIT))
      safe_offset = max(0, offset)
      normalized_session = _normalize_session_filter(session)
  
      original_query = query
      effective_query, was_corrected = _spell_correct(query)
  
      params: dict[str, Any] = {
          "q": effective_query,
          "mode": mode,
          "limit": capped_limit,
          "offset": safe_offset,
          "expand": expand,
          "has_answer": True,
      }
      if subject:
          params["subject"] = subject
      if paper is not None:
          params["paper"] = paper
      if year is not None:
          params["year"] = year
      if normalized_session:
          params["session"] = normalized_session
      if chapter is not None:
          params["chapter"] = chapter
  
      try:
          data = _api_get("/search", params)
      except Exception as exc:
          logger.error("search_questions failed: %s", exc, exc_info=True)
          error_payload = _error_from_exception(exc, "/search")
          message = error_payload.get("error", {}).get("message", "Search failed.")
          raise ToolError(message)
  
      raw_results = data.get("results", []) if isinstance(data, dict) else []
      if not isinstance(raw_results, list):
          raw_results = []
  
      visible_results = raw_results[:capped_limit]
      cards = [_result_item(r, i + 1) for i, r in enumerate(visible_results) if isinstance(r, dict)]
      all_result_ids = [r["id"] for r in cards if isinstance(r.get("id"), int)]
      recommended_ids = all_result_ids[: min(len(all_result_ids), MAX_RECOMMENDED_IDS)]
  
      payload = {
          "ok": True,
          "query": original_query,
          "effective_query": effective_query,
          "was_corrected": was_corrected,
          "filters": {
              "subject": subject,
              "paper": paper,
              "year": year,
              "session": normalized_session,
              "chapter": chapter,
              "mode": mode,
              "expand": expand,
              "limit": capped_limit,
              "offset": safe_offset,
          },
          "meta": {
              "total": data.get("total", len(raw_results)) if isinstance(data, dict) else len(raw_results),
              "returned": len(cards),
              "query_time_ms": data.get("query_time_ms") if isinstance(data, dict) else None,
          },
          "results": cards,
          "recommended_ids": recommended_ids,
          "next_step": {
              "tool": "get_questions",
              "question_ids_list": recommended_ids[:15],
              "example": "get_questions(question_ids_list=[1615,1684])",
          },
      }
  
      query_note = None
      if was_corrected and effective_query != original_query:
          query_note = f"Query corrected from '{original_query}' to '{effective_query}'."
  
      summary_text = _build_search_summary(
          title=f"Search results for '{original_query}'",
          query_note=query_note,
          total=payload["meta"]["total"],
          returned=payload["meta"]["returned"],
          cards=cards,
          recommended_ids=recommended_ids[:15],
      )
  
      return ToolResult(content=summary_text, structured_content=payload)

• mcp_server.py:229-237 (helper)

  Helper function used by search_questions to spell-correct the query before sending to the /search API endpoint.

  def _spell_correct(query: str) -> tuple[str, bool]:
      try:
          data = _api_get("/spellcheck", {"q": query})
          if isinstance(data, dict) and data.get("was_corrected") and data.get("corrected"):
              return str(data["corrected"]), True
      except Exception as exc:
          logger.debug("Spellcheck unavailable, continuing without correction: %s", exc)
      return query, False

• mcp_server.py:239-292 (helper)

  Helper function that transforms a raw API result row into a structured card item with paper identification, question metadata, image URLs, topic signal, and variant info. Called by search_questions to build each result entry.

  def _result_item(raw: dict[str, Any], index: int, matched_topics: Optional[list[str]] = None) -> dict[str, Any]:
      paper = raw.get("paper") or {}
      session_short = _short_session(paper.get("session_name"))
      paper_number = paper.get("paper_number")
      year = paper.get("year")
      variant = paper.get("variant")
  
      paper_label_parts: list[str] = []
      if paper_number is not None:
          paper_label_parts.append(f"P{paper_number}")
      if session_short and year is not None:
          paper_label_parts.append(f"{session_short}{year}")
      elif year is not None:
          paper_label_parts.append(str(year))
      if variant is not None:
          paper_label_parts.append(f"v{variant}")
  
      is_image_based = bool(raw.get("is_image_based"))
  
      item: dict[str, Any] = {
          "rank": index,
          "id": raw.get("id"),
          "paper": paper_number,
          "year": year,
          "session": session_short,
          "variant": variant,
          "paper_label": " ".join(paper_label_parts),
          "question_number": raw.get("question_number"),
          "marks": raw.get("marks"),
          "relevance_score": raw.get("relevance_score"),
          "snippet": _clean_text(raw.get("question_text"), max_len=200),
          "is_image_based": is_image_based,
          "question_url": f"{QUESTION_URL_BASE}/{raw.get('id')}" if raw.get("id") else None,
      }
  
      # Include image URLs for image-based questions
      if is_image_based:
          item["question_image_url"] = _to_image_url(raw.get("image_path"))
          item["ms_image_url"] = _to_image_url(raw.get("ms_image_path"))
  
      # Curated topic intelligence: just the one-line signal
      ti = raw.get("topic_intelligence")
      if isinstance(ti, dict) and ti.get("signal"):
          item["topic_signal"] = ti["signal"]
  
      # Variant info: count + IDs only, not full copies
      variants = raw.get("duplicate_variants", [])
      if isinstance(variants, list) and len(variants) > 0:
          item["variants_count"] = raw.get("duplicate_group_size", 1)
          item["variant_ids"] = [v.get("id") for v in variants if isinstance(v, dict) and v.get("id")]
  
      if matched_topics:
          item["matched_topics"] = matched_topics
      return item