search_policies

Instructions

Implementation Reference

• src/asf_policy_mcp/tools.py:62-125 (handler)

  The search_policies function is the handler for the tool. It searches cached ASF policy documents for a query term, ranks excerpts by relevance score, deduplicates nearby lines, and returns formatted results.

  @mcp.tool()
  def search_policies(query: str, max_results: int = 10) -> str:
      """Search across cached ASF policy documents for a query term or phrase.
  
      Returns ranked excerpts with surrounding context.
      Only searches policies already in the local cache — run refresh_cache first
      to ensure all policies are available.
      """
      if not query.strip():
          return "Please provide a search query."
  
      cache = fetcher.load_cache()
      query_words = set(query.lower().split())
      results: list[dict[str, Any]] = []
      skipped: list[str] = []
  
      for key in POLICY_SOURCES:
          entry = cache.get(key, {})
          text = str(entry.get("text", ""))
          if not text or text.startswith("[Error"):
              skipped.append(key)
              continue
          lines = text.split("\n")
          for i, line in enumerate(lines):
              score = sum(1 for w in query_words if w in line.lower())
              if score:
                  start = max(0, i - 2)
                  end = min(len(lines), i + 3)
                  results.append({
                      "key": key,
                      "title": POLICY_SOURCES[key]["title"],
                      "url": POLICY_SOURCES[key]["url"],
                      "score": score,
                      "excerpt": "\n".join(lines[start:end]).strip(),
                      "line": i,
                  })
  
      results.sort(key=lambda r: -r["score"])
  
      seen: dict[str, list[int]] = {}
      deduped: list[dict[str, Any]] = []
      for r in results:
          prior = seen.get(r["key"], [])
          if not any(abs(r["line"] - p) < 5 for p in prior):
              deduped.append(r)
              seen.setdefault(r["key"], []).append(r["line"])
          if len(deduped) >= max_results:
              break
  
      if not deduped:
          msg = f"No results found for **{query!r}**."
          if skipped:
              msg += f"\n\n_{len(skipped)} policies not yet cached were skipped. Run `refresh_cache` to fetch them._"
          return msg
  
      out = [f"# Search Results for '{query}'\n"]
      for r in deduped:
          out.append(f"## [{r['title']}]({r['url']})  (`{r['key']}`)\n")
          out.append("```")
          out.append(r["excerpt"])
          out.append("```\n")
      if skipped:
          out.append(f"_Note: {len(skipped)} uncached policies were not searched. Run `refresh_cache` to include them._")
      return "\n".join(out)

• src/asf_policy_mcp/tools.py:62-62 (registration)

  The @mcp.tool() decorator on line 62 registers search_policies as a FastMCP tool on the 'asf-policy' MCP server (mcp defined on line 14). No separate registration file exists; the decorator is the registration mechanism.

  @mcp.tool()

• src/asf_policy_mcp/tools.py:62-62 (schema)

  The schema is defined by the function signature: `query: str` (required) and `max_results: int = 10` (optional, default 10). The return type is `str`. FastMCP infers the JSON schema from these type annotations.

  @mcp.tool()