search

Instructions

Implementation Reference

• nexus_mcp.py:176-213 (handler)

  Handler function for the 'search' MCP tool. Reads all memory files, performs keyword matching on the content, extracts context snippets around matches, and returns formatted results with file name, tier, strength, and up to 3 snippets per match.

  def handle_tools_call(req_id, name, args):
      if name == "search":
          query = args.get("query", "").lower()
          if not query:
              return {"content": [{"type": "text", "text": "No query provided."}]}, None
          files = list_memory_files()
          results = []
          for f in files:
              with open(os.path.join(MEMORY_DIR, f["path"]), encoding="utf-8") as fh:
                  text = fh.read().lower()
              if query in text:
                  lines = text.splitlines()
                  snippets = []
                  for i, line in enumerate(lines):
                      if query in line.lower():
                          start = max(0, i - 1)
                          end = min(len(lines), i + 2)
                          snippet = "\n".join(lines[start:end]).strip()
                          if len(snippet) > 200:
                              snippet = snippet[:200] + "..."
                          snippets.append(snippet)
                  results.append({
                      "title": f["title"] or f["name"],
                      "path": f["path"],
                      "tier": f["tier"],
                      "strength": f["strength"],
                      "snippets": snippets[:3],
                  })
          if not results:
              return {"content": [{"type": "text", "text": f"No memories matched '{query}'."}]}, None
          lines = [f"Found {len(results)} memory(s) for '{query}':", ""]
          for r in results:
              lines.append(f"  [{r['tier']}] {r['title']}  (strength: {r['strength']:.2f})")
              lines.append(f"         path: {r['path']}")
              for s in r["snippets"]:
                  lines.append(f"         > {s}")
              lines.append("")
          return {"content": [{"type": "text", "text": "\n".join(lines)}]}, None

• nexus_mcp.py:121-129 (schema)

  Schema/definition of the 'search' tool. Declares the tool's name, description, and input schema requiring a 'query' string parameter.

  {
      "name": "search",
      "description": "Search across all memory files by keyword. Returns matching file names, titles, strengths, and context snippets.",
      "inputSchema": {
          "type": "object",
          "properties": {"query": {"type": "string", "description": "Keyword or phrase to search for"}},
          "required": ["query"],
      },
  },

• nexus_mcp.py:120-130 (registration)

  The 'search' tool is registered as part of the TOOL_DEFS list, which is returned by handle_tools_list() in response to a 'tools/list' MCP request.

  TOOL_DEFS = [
      {
          "name": "search",
          "description": "Search across all memory files by keyword. Returns matching file names, titles, strengths, and context snippets.",
          "inputSchema": {
              "type": "object",
              "properties": {"query": {"type": "string", "description": "Keyword or phrase to search for"}},
              "required": ["query"],
          },
      },
      {

• nexus_mcp.py:365-375 (registration)

  Dispatcher routing in the main loop: 'tools/list' calls handle_tools_list() which returns TOOL_DEFS (including 'search'), and 'tools/call' dispatches to handle_tools_call() where the 'search' case is handled.

  elif method == "tools/list":
      result = handle_tools_list(req_id)
      respond(req_id, result)
  elif method == "tools/call":
      name = params.get("name", "")
      args = params.get("arguments", {})
      result, err = handle_tools_call(req_id, name, args)
      if err:
          respond(req_id, error=err)
      else:
          respond(req_id, result)

• nexus_mcp.py:28-55 (helper)

  Helper function used by the search handler to enumerate all memory files across tiers with their metadata (name, title, tier, strength, path).

  def list_memory_files():
      """Return list of {tier, name, path, title, strength, type} for all memories."""
      results = []
      for tier in TIERS:
          tier_dir = os.path.join(MEMORY_DIR, tier)
          if not os.path.isdir(tier_dir):
              continue
          for f in sorted(glob.glob(os.path.join(tier_dir, "*.md"))):
              name = os.path.splitext(os.path.basename(f))[0]
              with open(f, encoding="utf-8") as fh:
                  content = fh.read()
              title = ""
              for line in content.splitlines():
                  if line.startswith("# "):
                      title = line[2:]
                      break
              strength = parse_frontmatter(content, "strength") or "1.0"
              type_ = parse_frontmatter(content, "type") or tier
              results.append({
                  "uri": f"nexus://{tier}/{name}",
                  "name": name,
                  "title": title,
                  "tier": tier,
                  "strength": float(strength),
                  "type": type_,
                  "path": os.path.relpath(f, MEMORY_DIR),
              })
      return results