get_chat_messages

Instructions

Implementation Reference

• src/imessage_mcp/server.py:33-44 (registration)

  Registration of 'get_chat_messages' as an MCP tool via @mcp.tool() decorator. The handler function delegates to db.get_chat_messages().

  @mcp.tool()
  def get_chat_messages(
      chat_id: int | None = None,
      handle: str | None = None,
      limit: int = 50,
      since: str | None = None,
  ) -> list[dict[str, Any]]:
      """Return messages for a chat. Provide chat_id or handle (phone/email).
  
      `since` is an ISO8601 datetime. Attachments are metadata only.
      """
      return db.get_chat_messages(chat_id=chat_id, handle=handle, limit=limit, since=since)

• src/imessage_mcp/db.py:162-212 (handler)

  Core implementation of get_chat_messages(). Builds SQL query to fetch messages from chat.db, filtered by chat_id or handle, with optional since filter. Returns messages oldest-first with sender, body, attachments.

  def get_chat_messages(
      chat_id: int | None = None,
      handle: str | None = None,
      limit: int = 50,
      since: str | None = None,
  ) -> list[dict[str, Any]]:
      if chat_id is None and not handle:
          raise ValueError("Provide chat_id or handle.")
      limit = max(1, min(int(limit), 500))
      with _open() as conn:
          params: list[Any] = []
          where: list[str] = []
          if chat_id is not None:
              join = "JOIN chat_message_join cmj ON cmj.message_id = m.ROWID"
              where.append("cmj.chat_id = ?")
              params.append(int(chat_id))
          else:
              join = "JOIN handle h ON h.ROWID = m.handle_id"
              where.append("LOWER(h.id) = LOWER(?)")
              params.append(normalize_handle(handle or ""))
          if since:
              where.append("m.date >= ?")
              params.append(iso_to_apple_ns(since))
          sql = f"""
              SELECT m.ROWID AS message_id, m.date, m.is_from_me, m.text, m.attributedBody,
                     m.handle_id,
                     (SELECT h2.id FROM handle h2 WHERE h2.ROWID = m.handle_id) AS sender_handle,
                     (SELECT m.cache_has_attachments) AS has_attach
              FROM message m
              {join}
              WHERE {' AND '.join(where)}
              ORDER BY m.date DESC
              LIMIT ?
          """
          params.append(limit)
          rows = conn.execute(sql, params).fetchall()
          out: list[dict[str, Any]] = []
          for r in rows:
              attachments = _attachments_for(conn, r["message_id"]) if r["has_attach"] else []
              out.append(
                  {
                      "message_id": r["message_id"],
                      "date": apple_ts_to_iso(r["date"]),
                      "is_from_me": bool(r["is_from_me"]),
                      "sender": None if r["is_from_me"] else r["sender_handle"],
                      "body": _extract_text(r),
                      "attachments": attachments,
                  }
              )
          out.reverse()  # return oldest-first for readability
          return out

• src/imessage_mcp/server.py:34-39 (schema)

  Input schema definition: chat_id (int|None), handle (str|None), limit (int, default 50), since (str|None ISO8601). Return type is list[dict].

  def get_chat_messages(
      chat_id: int | None = None,
      handle: str | None = None,
      limit: int = 50,
      since: str | None = None,
  ) -> list[dict[str, Any]]:

• src/imessage_mcp/handles.py:24-30 (helper)

  Helper 'iso_to_apple_ns' converts ISO8601 string to Apple nanosecond timestamp for the 'since' filter.

  def iso_to_apple_ns(iso_str: str) -> int:
      """Convert an ISO8601 string to Apple nanoseconds-since-2001-01-01."""
      dt = datetime.fromisoformat(iso_str.replace("Z", "+00:00"))
      if dt.tzinfo is None:
          dt = dt.replace(tzinfo=timezone.utc)
      unix_ts = dt.timestamp()
      return int((unix_ts - APPLE_EPOCH_OFFSET) * 1_000_000_000)

• src/imessage_mcp/handles.py:33-38 (helper)

  Helper 'normalize_handle' trims whitespace and lowercases email handles for SQL matching.

  def normalize_handle(value: str) -> str:
      """Trim whitespace. Keep + for phones, lowercase emails."""
      v = value.strip()
      if "@" in v:
          return v.lower()
      return v