Search Multiple Topics

Instructions

Implementation Reference

• mcp_server.py:635-639 (registration)

  Registration of the search_multi tool via @mcp.tool decorator with title 'Search Multiple Topics'

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

• mcp_server.py:640-826 (handler)

  Handler function `search_multi` that searches multiple topics via the API, deduplicates results, and returns a ToolResult with summary and structured content

  def search_multi(
      topics: str = "",
      topics_list: Optional[list[str]] = None,
      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_per_topic: int = 10,
      max_results: int = 40,
      expand: bool = True,
  ) -> ToolResult:
      """Search multiple topics and deduplicate by question ID.
  
      Accepts either `topics` (comma-separated string) or `topics_list`.
      """
      if not _validate_mode(mode):
          raise ToolError(
              "INVALID_MODE: mode must be one of 'hybrid', 'keyword', or 'semantic'."
          )
  
      topic_list = _parse_topics_input(topics, topics_list)
  
      if not topic_list:
          raise ToolError("NO_TOPICS: Provide one or more topics via topics or topics_list.")
      if len(topic_list) > MAX_TOPICS:
          raise ToolError(f"TOO_MANY_TOPICS: Maximum {MAX_TOPICS} topics per request.")
  
      capped_limit_per_topic = max(1, min(limit_per_topic, 30))
      capped_max_results = max(1, min(max_results, 100))
      normalized_session = _normalize_session_filter(session)
  
      all_results: dict[int, dict[str, Any]] = {}
      topic_breakdown: list[dict[str, Any]] = []
      effective_topics: list[str] = []
      effective_seen: set[str] = set()
  
      for topic in topic_list:
          corrected_topic, was_corrected = _spell_correct(topic)
          normalized_effective = _normalize_topic_key(corrected_topic)
  
          if normalized_effective in effective_seen:
              topic_breakdown.append(
                  {
                      "topic": topic,
                      "effective_topic": corrected_topic,
                      "was_corrected": was_corrected,
                      "api_returned": 0,
                      "new_unique_results": 0,
                      "skipped": "duplicate_effective_topic",
                  }
              )
              continue
  
          effective_seen.add(normalized_effective)
          effective_topics.append(corrected_topic)
  
          params: dict[str, Any] = {
              "q": corrected_topic,
              "mode": mode,
              "limit": capped_limit_per_topic,
              "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)
              result_rows = data.get("results", []) if isinstance(data, dict) else []
              if not isinstance(result_rows, list):
                  result_rows = []
  
              unique_added = 0
              for row in result_rows:
                  if not isinstance(row, dict):
                      continue
                  row_id = row.get("id")
                  if not isinstance(row_id, int):
                      continue
                  if row_id not in all_results:
                      all_results[row_id] = dict(row)
                      all_results[row_id]["_matched_topics"] = {corrected_topic}
                      unique_added += 1
                  else:
                      all_results[row_id].setdefault("_matched_topics", set()).add(corrected_topic)
  
              topic_breakdown.append(
                  {
                      "topic": topic,
                      "effective_topic": corrected_topic,
                      "was_corrected": was_corrected,
                      "api_returned": len(result_rows),
                      "new_unique_results": unique_added,
                  }
              )
          except Exception as exc:
              logger.warning("Topic search failed for '%s': %s", topic, exc)
              topic_breakdown.append(
                  {
                      "topic": topic,
                      "effective_topic": corrected_topic,
                      "was_corrected": was_corrected,
                      "error": str(exc),
                  }
              )
  
      merged = list(all_results.values())
      merged.sort(key=lambda x: float(x.get("relevance_score") or 0.0), reverse=True)
      visible = merged[:capped_max_results]
  
      cards: list[dict[str, Any]] = []
      for i, row in enumerate(visible, 1):
          matched_topics = sorted(list(row.get("_matched_topics", [])))
          cards.append(_result_item(row, i, matched_topics=matched_topics))
  
      all_result_ids = [r["id"] for r in cards if isinstance(r.get("id"), int)]
      recommended_ids = _select_recommended_ids(
          cards,
          limit=min(MAX_RECOMMENDED_IDS, len(all_result_ids)),
      )
  
      payload = {
          "ok": True,
          "topics": topic_list,
          "effective_topics": effective_topics,
          "filters": {
              "subject": subject,
              "paper": paper,
              "year": year,
              "session": normalized_session,
              "chapter": chapter,
              "mode": mode,
              "expand": expand,
              "limit_per_topic": capped_limit_per_topic,
              "max_results": capped_max_results,
          },
          "meta": {
              "topics_searched": len(topic_list),
              "unique_results_total": len(merged),
              "returned": len(cards),
          },
          "topic_breakdown": topic_breakdown,
          "results": cards,
          "recommended_ids": recommended_ids,
          "next_step": {
              "tool": "get_questions",
              "question_ids_list": recommended_ids,
              "example": "get_questions(question_ids_list=[1615,1684])",
          },
      }
  
      topic_lines: list[str] = []
      for row in topic_breakdown:
          if row.get("skipped"):
              topic_lines.append(
                  f"- '{row.get('topic')}': skipped (duplicate of '{row.get('effective_topic')}')."
              )
              continue
          if row.get("error"):
              topic_lines.append(f"- '{row.get('topic')}': error ({row.get('error')}).")
              continue
          topic_lines.append(
              f"- '{row.get('topic')}': {row.get('api_returned', 0)} found, "
              f"{row.get('new_unique_results', 0)} unique"
          )
  
      summary_text = _build_search_summary(
          title=f"Multi-topic search across {len(topic_list)} topics",
          query_note=None,
          total=payload["meta"]["unique_results_total"],
          returned=payload["meta"]["returned"],
          cards=cards,
          topic_lines=topic_lines,
          recommended_ids=recommended_ids,
      )
  
      return ToolResult(content=summary_text, structured_content=payload)

• mcp_server.py:307-308 (helper)

  Helper function `_parse_topics_input` used by search_multi to parse both comma-separated string and list-of-strings topic inputs

  def _parse_topics_input(topics: str, topics_list: Optional[list[str]]) -> list[str]:

• mcp_server.py:293-299 (helper)

  Helper function `_normalize_topic_key` used to normalize topic strings for deduplication

  def _validate_mode(mode: str) -> bool:
      return mode in {"hybrid", "keyword", "semantic"}
  
  
  def _normalize_topic_key(topic: str) -> str:

• mcp_server.py:447-471 (helper)

  Helper function `_select_recommended_ids` used by search_multi to select diverse recommended question IDs across topics

  def _select_recommended_ids(cards: list[dict[str, Any]], limit: int) -> list[int]:
      selected: list[int] = []
      covered_topics: set[str] = set()
  
      for card in cards:
          card_id = card.get("id")
          if not isinstance(card_id, int):
              continue
  
          card_topics = [str(t) for t in (card.get("matched_topics") or [])]
          unseen = [t for t in card_topics if t not in covered_topics]
          if unseen and card_id not in selected:
              selected.append(card_id)
              covered_topics.update(unseen)
              if len(selected) >= limit:
                  return selected[:limit]
  
      for card in cards:
          card_id = card.get("id")
          if isinstance(card_id, int) and card_id not in selected:
              selected.append(card_id)
              if len(selected) >= limit:
                  break
  
      return selected[:limit]