list_recent_chats

Instructions

Implementation Reference

• src/imessage_mcp/server.py:27-30 (handler)

  MCP tool handler that exposes list_recent_chats as a @mcp.tool(), delegating to db.list_recent_chats()

  @mcp.tool()
  def list_recent_chats(limit: int = 20) -> list[dict[str, Any]]:
      """List recent chats, newest-first, with previews and unread counts."""
      return db.list_recent_chats(limit=limit)

• src/imessage_mcp/server.py:27-30 (registration)

  The @mcp.tool() decorator registers list_recent_chats with the FastMCP server

  @mcp.tool()
  def list_recent_chats(limit: int = 20) -> list[dict[str, Any]]:
      """List recent chats, newest-first, with previews and unread counts."""
      return db.list_recent_chats(limit=limit)

• src/imessage_mcp/db.py:78-143 (handler)

  Core business logic: queries chat.db SQLite for recent chats (LIMIT clamped to 1-200), returns chat_id/display_name/participants/last_message_date/truncated preview/unread_count

  def list_recent_chats(limit: int = 20) -> list[dict[str, Any]]:
      limit = max(1, min(int(limit), 200))
      with _open() as conn:
          rows = conn.execute(
              """
              SELECT c.ROWID AS chat_id,
                     c.display_name,
                     c.chat_identifier,
                     (
                       SELECT GROUP_CONCAT(h.id, ', ')
                       FROM chat_handle_join chj
                       JOIN handle h ON h.ROWID = chj.handle_id
                       WHERE chj.chat_id = c.ROWID
                     ) AS participants,
                     (
                       SELECT MAX(m.date)
                       FROM chat_message_join cmj
                       JOIN message m ON m.ROWID = cmj.message_id
                       WHERE cmj.chat_id = c.ROWID
                     ) AS last_date,
                     (
                       SELECT COUNT(*)
                       FROM chat_message_join cmj
                       JOIN message m ON m.ROWID = cmj.message_id
                       WHERE cmj.chat_id = c.ROWID
                         AND m.is_read = 0
                         AND m.is_from_me = 0
                     ) AS unread_count
              FROM chat c
              WHERE EXISTS (
                  SELECT 1 FROM chat_message_join cmj WHERE cmj.chat_id = c.ROWID
              )
              ORDER BY last_date DESC
              LIMIT ?
              """,
              (limit,),
          ).fetchall()
  
          results: list[dict[str, Any]] = []
          for r in rows:
              msg_row = conn.execute(
                  """
                  SELECT m.text, m.attributedBody
                  FROM chat_message_join cmj
                  JOIN message m ON m.ROWID = cmj.message_id
                  WHERE cmj.chat_id = ?
                  ORDER BY m.date DESC
                  LIMIT 1
                  """,
                  (r["chat_id"],),
              ).fetchone()
              preview = _extract_text(msg_row) if msg_row else None
              if preview:
                  preview = preview[:120]
              display = r["display_name"] or r["chat_identifier"] or ""
              results.append(
                  {
                      "chat_id": r["chat_id"],
                      "display_name": display,
                      "participants": (r["participants"] or "").split(", ") if r["participants"] else [],
                      "last_message_date": apple_ts_to_iso(r["last_date"]),
                      "last_message_preview": preview,
                      "unread_count": r["unread_count"] or 0,
                  }
              )
          return results

• src/imessage_mcp/db.py:37-75 (helper)

  Helper used by list_recent_chats to extract message preview text from either the 'text' column or the attributedBody NSKeyedArchive blob

  def _extract_text(row: sqlite3.Row) -> str | None:
      """Return message.text, falling back to a best-effort read of attributedBody.
  
      attributedBody is an NSKeyedArchive blob. We do not parse it fully; we scan
      for the literal NSString payload that most text messages embed so that reply
      messages / richer content on newer macOS still show something useful.
      """
      text = row["text"]
      if text:
          return text
      blob: bytes | None = row["attributedBody"] if "attributedBody" in row.keys() else None
      if not blob:
          return None
      # typedstream layout after the NSString class tag:
      #   ... NSString <class-ref bytes> '+' <length-prefix> <utf-8 bytes>
      # The '+' (0x2b) byte is typedstream's variable-length-field marker.
      idx = blob.find(b"NSString")
      if idx == -1:
          return None
      plus = blob.find(b"+", idx)
      if plus == -1 or plus + 1 >= len(blob):
          return None
      cursor = plus + 1
      length_byte = blob[cursor]
      cursor += 1
      if length_byte == 0x81 and cursor + 2 <= len(blob):
          length = int.from_bytes(blob[cursor : cursor + 2], "little")
          cursor += 2
      elif length_byte == 0x82 and cursor + 4 <= len(blob):
          length = int.from_bytes(blob[cursor : cursor + 4], "little")
          cursor += 4
      elif length_byte < 0x80:
          length = length_byte
      else:
          return None
      try:
          return blob[cursor : cursor + length].decode("utf-8", errors="replace")
      except Exception:
          return None

• src/imessage_mcp/handles.py:9-21 (helper)

  Helper that converts Apple Core Data timestamps to ISO8601 strings, used by list_recent_chats for last_message_date

  def apple_ts_to_iso(apple_ts: int | None) -> str | None:
      """Convert Apple Core Data timestamp to ISO8601 UTC string.
  
      Newer macOS stores date as nanoseconds since 2001-01-01 UTC.
      Older rows stored plain seconds. Heuristic: values > 1e11 are nanoseconds.
      """
      if apple_ts is None or apple_ts == 0:
          return None
      if apple_ts > 10**11:
          unix_ts = apple_ts / 1_000_000_000 + APPLE_EPOCH_OFFSET
      else:
          unix_ts = apple_ts + APPLE_EPOCH_OFFSET
      return datetime.fromtimestamp(unix_ts, tz=timezone.utc).isoformat()