zettelkasten.zotero

zettelkasten.zotero

Zotero integration: Web API v3 client + local-install reader.

Two access paths:

• Web API (search/get_item/get_annotations) — metadata and annotations over the network, configured via ZOTERO_USER_ID and ZOTERO_API_KEY env vars, or a gitignored secrets/zotero_api file (env vars win). Cannot easily return file contents.
• Local install (fetch_pdf) — reads the desktop Zotero data directory (~/Zotero by default, override with ZOTERO_DATA_DIR) to resolve an item to its on-disk PDF, extract its text, and pull saved highlights/notes. No API key needed and works offline.

search

search(query: str) -> list[dict[str, Any]]

Search user's library items by title/creator.

Returns a list of items with key, title, authors, year, itemType, DOI.

Source code in zettelkasten/zotero.py

def search(query: str) -> list[dict[str, Any]]:
    """Search user's library items by title/creator.

    Returns a list of items with key, title, authors, year, itemType, DOI.
    """
    config = _get_config()
    if not config:
        return [{"error": _WEB_UNCONFIGURED_MSG}]

    user_id, api_key = config
    if not query.strip():
        return []

    resp = httpx.get(
        _base_url(user_id),
        headers=_headers(api_key),
        params={"q": query, "format": "json", "limit": 25},
        timeout=_TIMEOUT,
    )
    if resp.status_code != 200:
        return [{"error": f"Zotero API returned {resp.status_code}: {resp.text[:200]}"}]

    items = resp.json()
    results = []
    for item in items:
        data = item.get("data", {})
        creators = data.get("creators", [])
        authors = [
            f"{c.get('firstName', '')} {c.get('lastName', '')}".strip()
            for c in creators
            if c.get("creatorType") == "author"
        ]
        results.append({
            "key": item.get("key", ""),
            "title": data.get("title", ""),
            "authors": authors,
            "year": data.get("date", "")[:4],
            "itemType": data.get("itemType", ""),
            "DOI": data.get("DOI", ""),
        })
    return results

get_item

get_item(key: str) -> dict[str, Any]

Get full metadata for a single item by Zotero key.

Source code in zettelkasten/zotero.py

def get_item(key: str) -> dict[str, Any]:
    """Get full metadata for a single item by Zotero key."""
    config = _get_config()
    if not config:
        return {"error": _WEB_UNCONFIGURED_MSG, "type": "NotConfigured"}

    user_id, api_key = config
    if not key.strip():
        return {"error": "key must not be empty.", "type": "BadRequest"}

    resp = httpx.get(
        f"{_base_url(user_id)}/{key}",
        headers=_headers(api_key),
        params={"format": "json"},
        timeout=_TIMEOUT,
    )
    if resp.status_code != 200:
        return {"error": f"Zotero API returned {resp.status_code}: {resp.text[:200]}", "type": "UpstreamError"}

    item = resp.json()
    return item.get("data", item)

get_annotations

get_annotations(key: str) -> list[dict[str, Any]]

Get PDF annotations (highlights, notes) for an item's children.

Source code in zettelkasten/zotero.py

def get_annotations(key: str) -> list[dict[str, Any]]:
    """Get PDF annotations (highlights, notes) for an item's children."""
    config = _get_config()
    if not config:
        return [{"error": _WEB_UNCONFIGURED_MSG}]

    user_id, api_key = config
    if not key.strip():
        return []

    resp = httpx.get(
        f"{_base_url(user_id)}/{key}/children",
        headers=_headers(api_key),
        params={"format": "json"},
        timeout=_TIMEOUT,
    )
    if resp.status_code != 200:
        return [{"error": f"Zotero API returned {resp.status_code}: {resp.text[:200]}"}]

    children = resp.json()
    annotations = []
    for child in children:
        data = child.get("data", {})
        item_type = data.get("itemType", "")
        if item_type == "annotation":
            page_index, rect = _parse_annotation_position(
                data.get("annotationPosition")
            )
            annotations.append({
                "key": child.get("key", ""),
                "type": data.get("annotationType", ""),
                "text": data.get("annotationText", ""),
                "comment": data.get("annotationComment", ""),
                "color": data.get("annotationColor", ""),
                "page": data.get("annotationPageLabel", ""),
                # Location for image/area annotations (equation snapshots crop to
                # this rect). None on annotations without a parseable position.
                "page_index": page_index,
                "rect": rect,
            })
        elif item_type == "note":
            annotations.append({
                "key": child.get("key", ""),
                "type": "note",
                "text": data.get("note", ""),
                "comment": "",
                "color": "",
                "page": "",
            })
    return annotations

fetch_pdf

fetch_pdf(key: str, include_text: bool = True, include_annotations: bool = True, max_chars: int = 200000, full: bool = False) -> dict[str, Any]

Fetch a paper's full-text PDF from the local Zotero install.

Resolves a Zotero item key to its PDF attachment on disk and returns the absolute path plus (optionally) the extracted text and saved annotations.

When full is set the WHOLE document is extracted (ignoring max_chars for the returned/cached text) and the result carries page_offsets and a chapter outline so a caller can cache the complete book and address a later chapter. full bypasses the bounded LRU parse cache (that cache is keyed on the char cap and serves the Papers/Inspect prefix view).

Source code in zettelkasten/zotero.py

def fetch_pdf(
    key: str,
    include_text: bool = True,
    include_annotations: bool = True,
    max_chars: int = 200_000,
    full: bool = False,
) -> dict[str, Any]:
    """Fetch a paper's full-text PDF from the local Zotero install.

    Resolves a Zotero item key to its PDF attachment on disk and returns the
    absolute path plus (optionally) the extracted text and saved annotations.

    When ``full`` is set the WHOLE document is extracted (ignoring ``max_chars``
    for the returned/cached text) and the result carries ``page_offsets`` and a
    chapter ``outline`` so a caller can cache the complete book and address a
    later chapter. ``full`` bypasses the bounded LRU parse cache (that cache is
    keyed on the char cap and serves the Papers/Inspect prefix view).
    """
    key = (key or "").strip()
    if not key:
        return {"error": "key must not be empty.", "type": "BadRequest"}

    data_dir = _data_dir()
    db_path = data_dir / "zotero.sqlite"
    if not db_path.exists():
        return {
            "error": f"Zotero database not found at {db_path}. "
            "Set ZOTERO_DATA_DIR if your library lives elsewhere.",
            "type": "NotFound",
        }

    try:
        conn = _connect_ro(db_path)
    except sqlite3.Error as exc:
        return {"error": f"Could not open Zotero database: {exc}", "type": "IOError"}

    try:
        att = _resolve_pdf_attachment(conn, key)
        if att is None:
            return {"error": f"No PDF attachment found for item key {key!r}.", "type": "NotFound"}

        pdf_path = _attachment_file_path(data_dir, att)
        if pdf_path is None or not pdf_path.exists():
            mode = _LINK_MODE_LABELS.get(att["link_mode"], str(att["link_mode"]))
            return {
                "error": f"PDF for {key!r} is not available locally "
                f"(link mode {mode}; path {att['path']!r}).",
                "type": "NotFound",
            }

        result: dict[str, Any] = {
            "key": key,
            "path": str(pdf_path),
            "filename": pdf_path.name,
        }
        if include_text:
            if full:
                text, num_pages, warning, page_offsets, outline = (
                    _extract_pdf_text_full(pdf_path)
                )
                result["page_offsets"] = page_offsets
                result["outline"] = outline
            else:
                text, num_pages, warning = _extract_pdf_text_cached(
                    pdf_path, max_chars
                )
            result["text"] = text
            result["num_pages"] = num_pages
            if warning:
                result["text_warning"] = warning
        if include_annotations:
            result["annotations"] = _local_annotations(conn, att["attachment_key"])
        return result
    finally:
        conn.close()

find_item

find_item(doi: str = '', title: str = '', year: Any = None) -> dict[str, Any]

Resolve a work to an EXISTING Zotero item key, or report it's not there.

Checks whether a paper already lives in the user's Zotero library by matching its DOI (preferred) or title. Prefers the local install (zotero.sqlite, consistent with :func:fetch_pdf's PDF source) and falls back to the Web API when there's no local database. year is accepted for future tie-breaking but unused today.

Returns:

Type	Description
dict[str, Any]	One of three dicts: {"key": "<item key>", "source": "local"|"web"} when
dict[str, Any]	found; {"key": ""} when Zotero was searchable but the work isn't in it;
dict[str, Any]	or {"error": ..., "unconfigured": True} when there is no local DB and no
dict[str, Any]	Web API credentials, so membership cannot be determined.

Source code in zettelkasten/zotero.py

def find_item(doi: str = "", title: str = "", year: Any = None) -> dict[str, Any]:
    """Resolve a work to an EXISTING Zotero item key, or report it's not there.

    Checks whether a paper already lives in the user's Zotero library by matching
    its DOI (preferred) or title. Prefers the local install (``zotero.sqlite``,
    consistent with :func:`fetch_pdf`'s PDF source) and falls back to the Web API
    when there's no local database. ``year`` is accepted for future tie-breaking
    but unused today.

    Returns:
        One of three dicts: ``{"key": "<item key>", "source": "local"|"web"}`` when
        found; ``{"key": ""}`` when Zotero was searchable but the work isn't in it;
        or ``{"error": ..., "unconfigured": True}`` when there is no local DB and no
        Web API credentials, so membership cannot be determined.
    """
    doi_n = _normalize_doi(doi)
    title = (title or "").strip()

    local_searched = False
    db_path = _data_dir() / "zotero.sqlite"
    if db_path.exists():
        try:
            conn = _connect_ro(db_path)
            try:
                key = _local_find_key(conn, doi=doi_n, title=title)
            finally:
                conn.close()
            local_searched = True
            if key:
                return {"key": key, "source": "local"}
        except sqlite3.Error:
            local_searched = False

    if _get_config() is not None:
        key = _web_find_key(doi=doi_n, title=title)
        if key:
            return {"key": key, "source": "web"}
        return {"key": ""}

    if local_searched:
        return {"key": ""}

    return {"error": _UNCONFIGURED_MSG, "unconfigured": True, "type": "NotConfigured"}

list_collections

list_collections() -> dict[str, Any]

List the library's collections (folders), local-first with Web fallback.

Gives an agent a way to BROWSE the library structure — the themes/folders a user actually organizes their papers into — instead of only guessing at keyword searches. Prefers the local install (offline, no rate limits, same source as :func:fetch_pdf) and falls back to the Web API.

Source code in zettelkasten/zotero.py

def list_collections() -> dict[str, Any]:
    """List the library's collections (folders), local-first with Web fallback.

    Gives an agent a way to BROWSE the library structure — the themes/folders a
    user actually organizes their papers into — instead of only guessing at
    keyword searches. Prefers the local install (offline, no rate limits, same
    source as :func:`fetch_pdf`) and falls back to the Web API.
    """
    db_path = _data_dir() / "zotero.sqlite"
    if db_path.exists():
        try:
            conn = _connect_ro(db_path)
            try:
                cols = _local_collections(conn)
            finally:
                conn.close()
            return {"collections": cols, "count": len(cols), "source": "local"}
        except sqlite3.Error:
            pass  # fall through to Web API

    web = _web_collections()
    if web is None:
        return {"error": _UNCONFIGURED_MSG, "unconfigured": True, "type": "NotConfigured"}
    return {"collections": web, "count": len(web), "source": "web"}

list_items

list_items(collection: str = '', limit: int = 200) -> dict[str, Any]

List the papers in a collection (or the whole library), local-first.

collection accepts a collection KEY (preferred, stable) or its NAME (case-insensitive); empty lists the most recently added top-level items across the whole library. Returns search-shaped summaries (key, title, authors, year, itemType, DOI) ready to feed to :func:fetch_pdf.

Source code in zettelkasten/zotero.py

def list_items(collection: str = "", limit: int = 200) -> dict[str, Any]:
    """List the papers in a collection (or the whole library), local-first.

    ``collection`` accepts a collection KEY (preferred, stable) or its NAME
    (case-insensitive); empty lists the most recently added top-level items
    across the whole library. Returns search-shaped summaries (key, title,
    authors, year, itemType, DOI) ready to feed to :func:`fetch_pdf`.
    """
    collection = (collection or "").strip()
    db_path = _data_dir() / "zotero.sqlite"
    if db_path.exists():
        try:
            conn = _connect_ro(db_path)
            try:
                items = _local_collection_items(conn, collection, limit)
            finally:
                conn.close()
            if items is None:
                return {"error": f"Collection {collection!r} not found.", "type": "NotFound"}
            return {
                "collection": collection,
                "items": items,
                "count": len(items),
                "source": "local",
            }
        except sqlite3.Error:
            pass  # fall through to Web API

    web = _web_collection_items(collection, limit)
    if web is None:
        return {"error": _UNCONFIGURED_MSG, "unconfigured": True, "type": "NotConfigured"}
    if isinstance(web, dict) and web.get("not_found"):
        return {"error": f"Collection {collection!r} not found.", "type": "NotFound"}
    return {"collection": collection, "items": web, "count": len(web), "source": "web"}