whatsapp

whatsapp

WhatsApp chat export connector.

Reads WhatsApp chat history from exported .txt files. WhatsApp allows exporting individual chats via Chat → More → Export Chat on iOS/Android, producing plain-text files in a well-known format.

Export line format::

1/15/24, 10:30 AM - Alice: Hey, how are you?
1/15/24, 10:31 AM - Bob: Good thanks! Working on the project.

Each .txt file in the configured directory is parsed and yielded as a single :class:~openjarvis.connectors._stubs.Document with doc_type "message" and the message lines joined as content.

Classes

WhatsAppConnector

WhatsAppConnector(export_path: str = '')

Bases: BaseConnector

Read WhatsApp message history from exported .txt files.

WhatsApp allows exporting individual chats as plain-text files via Chat → More → Export Chat. Point this connector at the directory containing those files (one file per chat) and it will parse each file into a :class:~openjarvis.connectors._stubs.Document.

PARAMETER	DESCRIPTION
export_path	Path to the directory containing WhatsApp export .txt files. If empty, no documents are yielded. TYPE: str DEFAULT: ''

Source code in src/openjarvis/connectors/whatsapp.py

def __init__(self, export_path: str = "") -> None:
    self._export_path = Path(export_path) if export_path else Path("")
    self._items_synced: int = 0
    self._items_total: int = 0
    self._last_sync: Optional[datetime] = None

Functions

is_connected

is_connected() -> bool

Return True if the export directory exists and contains .txt files.

Source code in src/openjarvis/connectors/whatsapp.py

def is_connected(self) -> bool:
    """Return ``True`` if the export directory exists and contains .txt files."""
    if not self._export_path or not self._export_path.is_dir():
        return False
    return any(self._export_path.glob("*.txt"))

disconnect

disconnect() -> None

No-op: filesystem connector has nothing to disconnect.

Source code in src/openjarvis/connectors/whatsapp.py

def disconnect(self) -> None:
    """No-op: filesystem connector has nothing to disconnect."""

sync

sync(*, since: Optional[datetime] = None, cursor: Optional[str] = None) -> Iterator[Document]

Parse WhatsApp export files and yield one Document per chat file.

PARAMETER	DESCRIPTION
since	If provided, skip files (chats) whose most recent message is before this datetime. A file is included if any of its messages are on or after since. TYPE: Optional[datetime] DEFAULT: None
cursor	Not used (included for API compatibility). TYPE: Optional[str] DEFAULT: None

YIELDS	DESCRIPTION
Document	One document per .txt export file. The document content is the full parsed text of the chat with timestamp and sender prefixes. Participants are extracted from the set of senders found.

Source code in src/openjarvis/connectors/whatsapp.py

def sync(
    self,
    *,
    since: Optional[datetime] = None,
    cursor: Optional[str] = None,  # noqa: ARG002
) -> Iterator[Document]:
    """Parse WhatsApp export files and yield one Document per chat file.

    Parameters
    ----------
    since:
        If provided, skip files (chats) whose most recent message is
        before this datetime.  A file is included if *any* of its
        messages are on or after *since*.
    cursor:
        Not used (included for API compatibility).

    Yields
    ------
    Document
        One document per .txt export file.  The document content is the
        full parsed text of the chat with timestamp and sender prefixes.
        Participants are extracted from the set of senders found.
    """
    if not self._export_path or not self._export_path.is_dir():
        return

    txt_files = sorted(self._export_path.glob("*.txt"))
    self._items_total = len(txt_files)
    synced = 0

    since_utc: Optional[datetime] = None
    if since is not None:
        if since.tzinfo is None:
            since_utc = since.replace(tzinfo=timezone.utc)
        else:
            since_utc = since

    for txt_file in txt_files:
        doc = self._parse_export_file(txt_file, since=since_utc)
        if doc is not None:
            synced += 1
            yield doc

    self._items_synced = synced
    self._last_sync = datetime.now(tz=timezone.utc)

sync_status

sync_status() -> SyncStatus

Return sync progress from the most recent :meth:sync call.

Source code in src/openjarvis/connectors/whatsapp.py

def sync_status(self) -> SyncStatus:
    """Return sync progress from the most recent :meth:`sync` call."""
    return SyncStatus(
        state="idle",
        items_synced=self._items_synced,
        items_total=self._items_total,
        last_sync=self._last_sync,
    )

mcp_tools

mcp_tools() -> List[ToolSpec]

Expose two MCP tool specs for real-time WhatsApp queries.

Source code in src/openjarvis/connectors/whatsapp.py

def mcp_tools(self) -> List[ToolSpec]:
    """Expose two MCP tool specs for real-time WhatsApp queries."""
    return [
        ToolSpec(
            name="whatsapp_search_messages",
            description=(
                "Search WhatsApp messages by keyword or sender name. "
                "Returns matching messages with sender and timestamp."
            ),
            parameters={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Search query string",
                    },
                    "max_results": {
                        "type": "integer",
                        "description": "Maximum number of messages to return",
                        "default": 20,
                    },
                },
                "required": ["query"],
            },
            category="knowledge",
        ),
        ToolSpec(
            name="whatsapp_get_chat",
            description=(
                "Retrieve the full message history for a specific WhatsApp chat "
                "by chat name (the export filename without the .txt extension)."
            ),
            parameters={
                "type": "object",
                "properties": {
                    "chat_name": {
                        "type": "string",
                        "description": (
                            "Name of the chat (matches the .txt filename, "
                            "e.g. 'WhatsApp Chat with Alice')"
                        ),
                    },
                    "max_results": {
                        "type": "integer",
                        "description": "Maximum number of messages to return",
                        "default": 50,
                    },
                },
                "required": ["chat_name"],
            },
            category="knowledge",
        ),
    ]