apple_contacts

apple_contacts

Apple Contacts connector — reads directly from the macOS Contacts SQLite database.

No API calls, no OAuth. The connector opens ~/Library/Application Support/AddressBook/AddressBook-v22.abcddb in read-only mode and yields one :class:Document per contact.

Requires Full Disk Access granted to the terminal / app in System Settings → Privacy & Security → Full Disk Access.

Timestamp notes

The Contacts database stores timestamps as seconds since the Apple epoch of 2001-01-01 00:00:00 UTC.

Classes

AppleContactsConnector

AppleContactsConnector(db_path: str = '')

Bases: BaseConnector

Connector that reads contacts from the macOS Contacts SQLite database.

PARAMETER	DESCRIPTION
db_path	Path to AddressBook-v22.abcddb. Defaults to ~/Library/Application Support/AddressBook/AddressBook-v22.abcddb. TYPE: str DEFAULT: ''

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

def __init__(self, db_path: str = "") -> None:
    self._db_path: Path = Path(db_path) if db_path else _DEFAULT_DB_PATH
    self._connected: bool = False
    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 any AddressBook database exists.

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

def is_connected(self) -> bool:
    """Return ``True`` if any AddressBook database exists."""
    return len(self._all_db_paths()) > 0

disconnect

disconnect() -> None

Mark the connector as disconnected.

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

def disconnect(self) -> None:
    """Mark the connector as disconnected."""
    self._connected = False

sync

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

Read contacts from AddressBook and yield one :class:Document each.

PARAMETER	DESCRIPTION
since	If provided, skip contacts whose modification time is before this datetime. TYPE: Optional[datetime] DEFAULT: None
cursor	Not used for this local connector (included for API compatibility). TYPE: Optional[str] DEFAULT: None

YIELDS	DESCRIPTION
Document	One document per contact with all fields as structured text.

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

def sync(
    self,
    *,
    since: Optional[datetime] = None,
    cursor: Optional[str] = None,  # noqa: ARG002
) -> Iterator[Document]:
    """Read contacts from AddressBook and yield one :class:`Document` each.

    Parameters
    ----------
    since:
        If provided, skip contacts whose modification time is before
        this datetime.
    cursor:
        Not used for this local connector (included for API
        compatibility).

    Yields
    ------
    Document
        One document per contact with all fields as structured text.
    """
    db_paths = self._all_db_paths()
    if not db_paths:
        return

    seen_ids: set[str] = set()
    synced = 0
    total = 0

    for db_path in db_paths:
        conn = self._open_db(db_path)
        if conn is None:
            continue

        try:
            rows = conn.execute(_CONTACTS_QUERY).fetchall()
            total += len(rows)
            self._items_total = total

            for row in rows:
                pk: int = row[0]
                first: str = row[1] or ""
                middle: str = row[2] or ""
                last: str = row[3] or ""
                org: str = row[4] or ""
                job_title: str = row[5] or ""
                department: str = row[6] or ""
                nickname: str = row[7] or ""
                birthday_ts: float | None = row[8]
                creation_ts: float = row[9] or 0.0
                mod_ts: float = row[10] or 0.0
                unique_id: str = row[11] or str(pk)

                # Deduplicate across sources
                if unique_id in seen_ids:
                    continue
                seen_ids.add(unique_id)

                timestamp = (
                    _apple_ts_to_datetime(mod_ts)
                    if mod_ts
                    else _apple_ts_to_datetime(creation_ts)
                )

                # Apply since filter
                if since is not None:
                    since_utc = since
                    if since_utc.tzinfo is None:
                        since_utc = since_utc.replace(tzinfo=timezone.utc)
                    if timestamp < since_utc:
                        continue

                name = _build_name(first, middle, last, org)
                birthday = ""
                if birthday_ts:
                    try:
                        birthday = _apple_ts_to_datetime(birthday_ts).strftime(
                            "%Y-%m-%d"
                        )
                    except (ValueError, OverflowError):
                        pass

                meta: Dict[str, Any] = {
                    "name": name,
                    "first_name": first,
                    "last_name": last,
                    "organization": org,
                    "job_title": job_title,
                    "department": department,
                    "nickname": nickname,
                    "birthday": birthday,
                }

                content = self._build_content(conn, pk, meta)

                doc = Document(
                    doc_id=f"apple_contacts:{unique_id}",
                    source="apple_contacts",
                    doc_type="contact",
                    content=content,
                    title=name,
                    author=name,
                    timestamp=timestamp,
                    metadata=meta,
                )
                synced += 1
                yield doc

        finally:
            conn.close()

    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/apple_contacts.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 MCP tool specs for real-time Apple Contacts queries.

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

def mcp_tools(self) -> List[ToolSpec]:
    """Expose MCP tool specs for real-time Apple Contacts queries."""
    return [
        ToolSpec(
            name="contacts_search",
            description=(
                "Search Apple Contacts by name, organization, or job title. "
                "Returns matching contacts with full details."
            ),
            parameters={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Search query string",
                    },
                    "max_results": {
                        "type": "integer",
                        "description": "Maximum number of contacts to return",
                        "default": 20,
                    },
                },
                "required": ["query"],
            },
            category="knowledge",
        ),
        ToolSpec(
            name="contacts_get_contact",
            description=(
                "Retrieve full details of an Apple Contact by its "
                "unique identifier."
            ),
            parameters={
                "type": "object",
                "properties": {
                    "contact_id": {
                        "type": "string",
                        "description": (
                            "Apple Contacts unique identifier (ZUNIQUEID)"
                        ),
                    },
                },
                "required": ["contact_id"],
            },
            category="knowledge",
        ),
    ]