recommend_next

Instructions

Implementation Reference

• src/kothar/server.py:77-116 (handler)

  The main handler function for the 'recommend_next' MCP tool. It accepts current_stack (list of server names), new_context (description of new needs), and an optional session_file path. It ensures the search index is ready, optionally reads session notes from a file to append to context, then calls find_similar with exclude filtering and formats results.

  @mcp.tool()
  def recommend_next(
      current_stack: list[str],
      new_context: str,
      session_file: str | None = None,
  ) -> str:
      """
      Mid-project advisor: given your current MCP stack (list of server names) and a new
      development context, recommend what to add next and why.
      Example: current_stack=["github", "filesystem"], new_context="adding Stripe payments and PDF invoices"
      session_file: optional path to a session notes file whose content is appended to new_context.
      """
      try:
          _ensure_index()
          installed = [s.strip() for s in current_stack if s.strip()]
  
          context = new_context
          if session_file:
              try:
                  with open(session_file) as f:
                      session_content = f.read().strip()
                  if session_content:
                      context = f"{new_context}\n\n{session_content}"
              except OSError as e:
                  return (
                      f"## Error reading session file\n\n"
                      f"Could not read `{session_file}`: {e}\n\n"
                      f"Fix the path or omit `session_file` to proceed without it."
                  )
  
          results = find_similar(context, top_k=5, exclude=installed)
  
          header = (
              f"## What to add next\n"
              f"**Current stack:** {current_stack}\n"
              f"**New context:** {new_context}\n\n"
          )
          return header + _format_results(results, context)
      except Exception as e:
          return _error_response("computing next recommendations", e)

• src/kothar/server.py:77-77 (registration)

  The @mcp.tool() decorator from FastMCP registers 'recommend_next' as an MCP tool on the server named 'kothar'.

  @mcp.tool()

• src/kothar/search.py:58-114 (helper)

  Core semantic search helper called by recommend_next. Encodes the query, performs cosine similarity search against the DuckDB index, filters out excluded servers (by name), and has an adaptive fallback to return closest results even below threshold.

  def find_similar(
      query: str,
      top_k: int = 10,
      exclude: list[str] | None = None,
      min_score: float = DEFAULT_MIN_SCORE,
  ) -> list[dict]:
      """
      Return top_k servers most semantically similar to query.
      Each result: {name, url, description, category, score}
      """
      model = _get_model()
      query_emb = _encode_query(model, query)
  
      excluded_lower = {n.lower() for n in (exclude or [])}
  
      con = get_connection()
      rows = con.execute(
          """
          SELECT name, description, url, category, score FROM (
              SELECT name, description, url, category,
                     array_cosine_similarity(embedding, ?::FLOAT[384]) AS score
              FROM servers
          )
          WHERE score >= ?
          ORDER BY score DESC
          """,
          [query_emb, min_score],
      ).fetchall()
      con.close()
  
      results = []
      for name, desc, url, cat, score in rows:
          name_lower = name.lower()
          short_name = name_lower.split("/")[-1]
          if any(
              e == name_lower or e == short_name or e in name_lower
              for e in excluded_lower
          ):
              continue
          results.append(
              {
                  "name": name,
                  "url": url,
                  "description": desc,
                  "category": cat,
                  "score": float(score),
              }
          )
          if len(results) >= top_k:
              break
  
      # Adaptive fallback: if nothing cleared the threshold, return the closest
      # matches below it so callers never silently get zero results.
      if not results and min_score > 0:
          return find_similar(query, top_k=top_k, exclude=exclude, min_score=0.0)
  
      return results

• src/kothar/server.py:47-59 (helper)

  Helper that formats the list of server results into Markdown, generating a rationale for each via generate_rationale.

  def _format_results(results: list[dict], project_description: str) -> str:
      if not results:
          return "No matching MCP servers found. Try a more descriptive project description."
  
      lines = []
      for i, r in enumerate(results, 1):
          rationale = generate_rationale(r, project_description)
          lines.append(
              f"{i}. **{r['name']}**\n"
              f"   {r['url']}\n"
              f"   {rationale}\n"
          )
      return "\n".join(lines)

• src/kothar/server.py:22-199 (registration)

  The FastMCP server is instantiated with name 'kothar' and runs via main(); this is the server context in which @mcp.tool() decorators register tools.

  mcp = FastMCP("kothar")
  
  _index_initialized = False
  
  
  def _ensure_index() -> None:
      global _index_initialized
      if _index_initialized:
          return
      if not is_index_ready():
          print("Index not found — building now (first run, ~30s)...", file=sys.stderr)
          build_index()
      _index_initialized = True
  
  
  def _error_response(context: str, exc: Exception) -> str:
      traceback.print_exc(file=sys.stderr)
      return (
          f"## Error while {context}\n\n"
          f"{type(exc).__name__}: {exc}\n\n"
          f"See server logs for details. If this persists, try rebuilding the "
          f"index with `uv run python -m kothar.indexer --force`."
      )
  
  
  def _format_results(results: list[dict], project_description: str) -> str:
      if not results:
          return "No matching MCP servers found. Try a more descriptive project description."
  
      lines = []
      for i, r in enumerate(results, 1):
          rationale = generate_rationale(r, project_description)
          lines.append(
              f"{i}. **{r['name']}**\n"
              f"   {r['url']}\n"
              f"   {rationale}\n"
          )
      return "\n".join(lines)
  
  
  @mcp.tool()
  def recommend_for_project(description: str) -> str:
      """
      Given a project description, recommend the top MCP servers to install and explain why each one fits.
      Example: "Python FastAPI backend with PostgreSQL and JWT auth"
      """
      try:
          _ensure_index()
          results = find_similar(description, top_k=5)
          header = f"## Recommended MCP servers for: {description}\n\n"
          return header + _format_results(results, description)
      except Exception as e:
          return _error_response("generating recommendations", e)
  
  
  @mcp.tool()
  def recommend_next(
      current_stack: list[str],
      new_context: str,
      session_file: str | None = None,
  ) -> str:
      """
      Mid-project advisor: given your current MCP stack (list of server names) and a new
      development context, recommend what to add next and why.
      Example: current_stack=["github", "filesystem"], new_context="adding Stripe payments and PDF invoices"
      session_file: optional path to a session notes file whose content is appended to new_context.
      """
      try:
          _ensure_index()
          installed = [s.strip() for s in current_stack if s.strip()]
  
          context = new_context
          if session_file:
              try:
                  with open(session_file) as f:
                      session_content = f.read().strip()
                  if session_content:
                      context = f"{new_context}\n\n{session_content}"
              except OSError as e:
                  return (
                      f"## Error reading session file\n\n"
                      f"Could not read `{session_file}`: {e}\n\n"
                      f"Fix the path or omit `session_file` to proceed without it."
                  )
  
          results = find_similar(context, top_k=5, exclude=installed)
  
          header = (
              f"## What to add next\n"
              f"**Current stack:** {current_stack}\n"
              f"**New context:** {new_context}\n\n"
          )
          return header + _format_results(results, context)
      except Exception as e:
          return _error_response("computing next recommendations", e)
  
  
  @mcp.tool()
  def explain_why(server_name: str, project_description: str) -> str:
      """
      Explain why a specific MCP server is a good fit for a given project.
      Example: server_name="github", project_description="open source Python library with CI/CD"
      """
      try:
          _ensure_index()
          server = lookup_by_name(server_name)
          if server is None:
              return (
                  f"Could not find '{server_name}' in the index. "
                  f"Try a partial name or check spelling."
              )
  
          rationale = generate_rationale(server, project_description)
          return (
              f"## Why {server['name']} fits your project\n\n"
              f"**Server:** {server['name']}\n"
              f"**URL:** {server['url']}\n"
              f"**Category:** {server['category']}\n"
              f"**Description:** {server['description']}\n\n"
              f"**Rationale:** {rationale}"
          )
      except Exception as e:
          return _error_response(f"explaining fit for '{server_name}'", e)
  
  
  # Splits on `. `, `; `, ` and then `, `, then `, ` then ` — NOT bare ` and `.
  # Longer alternatives listed first so regex tries them before shorter overlaps.
  _GOAL_SPLIT_RE = re.compile(
      r"\s+and\s+then\s+|,\s+then\s+|\s+then\s+|\.\s+|;\s+",
      re.IGNORECASE,
  )
  
  
  def _split_goal(goal: str) -> list[str]:
      return [p.strip() for p in _GOAL_SPLIT_RE.split(goal) if p.strip()]
  
  
  @mcp.tool()
  def recommend_for_goal(goal: str, project: str | None = None) -> str:
      """
      Decompose a multi-part goal into sub-queries and recommend MCP servers for each part.
      Splits on hard boundaries: '. ', '; ', ' then ', ', then ', ' and then ' (not bare ' and ').
      project: optional project context prepended to each sub-query for richer semantic matching.
      Example: goal="integrate GitHub. add Stripe payments", project="Python FastAPI backend"
      """
      try:
          _ensure_index()
  
          if not goal or not goal.strip():
              return "Please provide a goal description."
  
          parts = _split_goal(goal)
          seen_names: set[str] = set()
  
          if len(parts) == 1:
              query = f"{project}\n\n{parts[0]}" if project else parts[0]
              results = find_similar(query, top_k=5)
              header = f"## Recommended MCP servers for: {goal}\n\n"
              return header + _format_results(results, query)
  
          sections: list[str] = []
          for part in parts:
              query = f"{project}\n\n{part}" if project else part
              results = find_similar(query, top_k=5)
              fresh = [r for r in results if r["name"] not in seen_names]
              seen_names.update(r["name"] for r in fresh)
              sections.append(f"### {part}\n\n{_format_results(fresh, query)}")
  
          return "\n\n".join(sections)
      except Exception as e:
          return _error_response("computing goal recommendations", e)
  
  
  def main() -> None:
      """Entry point for the `kothar` console script (see pyproject [project.scripts])."""
      mcp.run()