recommend_for_goal

Instructions

Implementation Reference

• src/kothar/server.py:160-192 (handler)

  The main tool handler: receives a goal string, decomposes it via _split_goal into sub-queries, deduplicates results by server name, and returns grouped recommendations. Supports an optional project context parameter for richer semantic matching.

  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)

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

  The @mcp.tool() decorator registers recommend_for_goal as a FastMCP tool with the name 'recommend_for_goal'.

  @mcp.tool()

• src/kothar/server.py:160-166 (schema)

  The function signature defines the input schema: goal (required str) and project (optional str). The docstring serves as the tool description.

  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"
      """

• src/kothar/server.py:149-156 (helper)

  _GOAL_SPLIT_RE is a compiled regex that splits goals on boundaries like '. ', '; ', ' then ', ', then ', ' and then ' (but NOT bare 'and'). _split_goal uses this regex to decompose the goal string into sub-query parts.

  _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()]

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

  find_similar is the core search function called by recommend_for_goal. It encodes the query, performs cosine similarity search via DuckDB, filters by score threshold, and returns top matching servers.

  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