search_imessages_rich

Instructions

Implementation Reference

• src/imessage_rich_search/mcp_server.py:94-138 (handler)

  Main tool call handler for 'search_imessages_rich'. Validates inputs (query, contact, limit), calls search() from cli.py, and returns results as JSON-RPC response.

  def _handle_tools_call(msg_id: Any, params: Dict[str, Any]) -> Dict[str, Any]:
      name = params.get("name")
      args = params.get("arguments") or {}
  
      if name != "search_imessages_rich":
          return _err(msg_id, -32602, f"unknown tool: {name!r}")
  
      query = args.get("query")
      if not isinstance(query, str) or not query:
          return _err(msg_id, -32602, "'query' is required and must be a non-empty string")
      contact = args.get("contact")
      if contact is not None and not isinstance(contact, str):
          return _err(msg_id, -32602, "'contact' must be a string if provided")
      limit = args.get("limit", 50)
      if not isinstance(limit, int) or limit < 1 or limit > 1000:
          return _err(msg_id, -32602, "'limit' must be an integer in [1, 1000]")
  
      try:
          results = [asdict(m) for m in search(query, contact, limit)]
      except FileNotFoundError as e:
          return _ok(msg_id, {
              "content": [{"type": "text", "text": f"error: {e}"}],
              "isError": True,
          })
      except Exception as e:
          return _ok(msg_id, {
              "content": [{
                  "type": "text",
                  "text": f"error: {type(e).__name__}: {e}\n{traceback.format_exc()}",
              }],
              "isError": True,
          })
  
      summary = (
          f"{len(results)} match(es) for {query!r}"
          + (f" (contact={contact})" if contact else "")
      )
      return _ok(msg_id, {
          "content": [
              {"type": "text", "text": summary},
              {"type": "text", "text": json.dumps(results, ensure_ascii=False, indent=2)},
          ],
          "structuredContent": {"matches": results, "count": len(results)},
          "isError": False,
      })

• src/imessage_rich_search/mcp_server.py:30-61 (schema)

  Tool registration and input schema for 'search_imessages_rich'. Defines name, description, and inputSchema with query (required string), contact (optional string), and limit (optional integer, default 50, max 1000).

  TOOLS = [
      {
          "name": "search_imessages_rich",
          "description": (
              "Full-text search across macOS iMessages including link preview metadata "
              "(title, summary, site name) that Apple stores in payload_data — content "
              "the basic chat.db text column does not expose. Returns newest-first matches "
              "where the query (case-insensitive substring) appears in either the message "
              "body OR the rich link preview."
          ),
          "inputSchema": {
              "type": "object",
              "properties": {
                  "query": {
                      "type": "string",
                      "description": "Substring to search for (case-insensitive).",
                  },
                  "contact": {
                      "type": "string",
                      "description": "Optional handle filter, e.g. '+14073993471' or 'name@example.com'.",
                  },
                  "limit": {
                      "type": "integer",
                      "description": "Max matches to return (default 50).",
                      "default": 50,
                      "minimum": 1,
                      "maximum": 1000,
                  },
              },
              "required": ["query"],
          },
      }

• src/imessage_rich_search/mcp_server.py:30-62 (registration)

  The TOOLS list containing the 'search_imessages_rich' tool definition, registered for the MCP tools/list endpoint.

  TOOLS = [
      {
          "name": "search_imessages_rich",
          "description": (
              "Full-text search across macOS iMessages including link preview metadata "
              "(title, summary, site name) that Apple stores in payload_data — content "
              "the basic chat.db text column does not expose. Returns newest-first matches "
              "where the query (case-insensitive substring) appears in either the message "
              "body OR the rich link preview."
          ),
          "inputSchema": {
              "type": "object",
              "properties": {
                  "query": {
                      "type": "string",
                      "description": "Substring to search for (case-insensitive).",
                  },
                  "contact": {
                      "type": "string",
                      "description": "Optional handle filter, e.g. '+14073993471' or 'name@example.com'.",
                  },
                  "limit": {
                      "type": "integer",
                      "description": "Max matches to return (default 50).",
                      "default": 50,
                      "minimum": 1,
                      "maximum": 1000,
                  },
              },
              "required": ["query"],
          },
      }
  ]

• src/imessage_rich_search/cli.py:69-127 (helper)

  Core search() function called by the MCP handler. Queries chat.db via SQLite, extracts preview metadata from NSKeyedArchiver bplist payload_data, and filters by case-insensitive substring match on combined text + preview.

  def search(
      query: str,
      contact: Optional[str] = None,
      limit: int = 200,
      db_path: Path = CHAT_DB,
  ) -> list:
      """Search messages where text OR extracted preview metadata contains `query`.
  
      Args:
          query: Case-insensitive substring to match.
          contact: Optional handle filter (e.g. '+14073993471' or 'foo@bar.com').
          limit: Maximum matches to return.
          db_path: Override path to chat.db (defaults to ~/Library/Messages/chat.db).
  
      Returns:
          List of Match objects, newest first.
      """
      if not db_path.exists():
          raise FileNotFoundError(
              f"chat.db not found at {db_path}. "
              "Grant Full Disk Access to your terminal in System Settings → "
              "Privacy & Security → Full Disk Access."
          )
      q = query.lower()
      conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
      conn.row_factory = sqlite3.Row
      cur = conn.cursor()
  
      sql = """
          SELECT m.ROWID, m.text, m.date, m.is_from_me,
                 m.payload_data, m.balloon_bundle_id, h.id AS handle
          FROM message m
          LEFT JOIN handle h ON m.handle_id = h.ROWID
      """
      params: list = []
      if contact:
          sql += " WHERE h.id = ?"
          params.append(contact)
      sql += " ORDER BY m.date DESC"
  
      matches: list = []
      for row in cur.execute(sql, params):
          text = row["text"] or ""
          preview = extract_strings(row["payload_data"])
          haystack = (text + "\n" + "\n".join(preview)).lower()
          if q in haystack:
              matches.append(Match(
                  rowid=row["ROWID"],
                  date=apple_ns_to_iso(row["date"]),
                  is_from_me=bool(row["is_from_me"]),
                  handle=row["handle"],
                  text=text,
                  preview=preview,
                  balloon=row["balloon_bundle_id"],
              ))
              if len(matches) >= limit:
                  break
      conn.close()
      return matches

• src/imessage_rich_search/cli.py:47-66 (helper)

  extract_strings() helper that parses NSKeyedArchiver binary plists from payload_data and extracts human-readable strings for rich link preview metadata search.

  def extract_strings(blob: Optional[bytes]) -> list:
      """Pull human-readable strings from an NSKeyedArchiver binary plist.
  
      We don't reconstruct the object graph — we just collect every string in
      the `$objects` array, which is sufficient for full-text search and avoids
      a heavy dependency like ccl_bplist or pyobjc.
      """
      if not blob:
          return []
      try:
          plist = plistlib.loads(blob)
      except Exception:
          return []
      out = []
      for o in plist.get("$objects", []):
          if isinstance(o, str) and o and o != "$null" and not o.startswith("$"):
              # Filter out short Foundation class names like 'NSURL', 'NSDate'
              if not (o.startswith("NS") and len(o) < 30):
                  out.append(o)
      return out