search_zotero_items

Instructions

Implementation Reference

• annota/server.py:188-199 (handler)

  The MCP tool handler for search_zotero_items. Decorated with @mcp.tool(), it accepts a query string and optional limit (default 20), then delegates to zotero_db.search_items() and returns JSON results.

  @mcp.tool()
  def search_zotero_items(query: str, limit: int = 20) -> str:
      """按标题、作者或 key 搜索 Zotero 条目。
  
      比 list_zotero_items 更高效，可直接定位目标论文。
  
      Args:
          query: 搜索关键词（标题/作者的部分文字，或 Zotero item key）
          limit: 最多返回条目数（默认 20）
      """
      items = zotero_db.search_items(query=query, limit=limit)
      return json.dumps(items, ensure_ascii=False, indent=2)

• annota/zotero_db.py:235-288 (handler)

  The actual implementation of search_items() in the Zotero database layer. It connects to the Zotero SQLite database (read-only copy) and performs a fuzzy search across item titles, creator last/first names, and item keys, returning a list of items with optional PDF attachment info.

  def search_items(query: str, limit: int = 20) -> list[dict]:
      """按标题或作者模糊搜索 Zotero 条目。"""
      conn = _connect(readonly=True)
      try:
          pattern = f"%{query}%"
          rows = conn.execute("""
              SELECT DISTINCT
                  i.itemID,
                  i.key,
                  idv_title.value AS title
              FROM items i
              LEFT JOIN itemData id_title
                  ON i.itemID = id_title.itemID
                  AND id_title.fieldID = (
                      SELECT fieldID FROM fields WHERE fieldName = 'title'
                  )
              LEFT JOIN itemDataValues idv_title
                  ON id_title.valueID = idv_title.valueID
              LEFT JOIN itemCreators ic ON i.itemID = ic.itemID
              LEFT JOIN creators c ON ic.creatorID = c.creatorID
              WHERE i.itemTypeID NOT IN (1, 3, 28)
                  AND i.libraryID = ?
                  AND (
                      idv_title.value LIKE ?
                      OR c.lastName LIKE ?
                      OR c.firstName LIKE ?
                      OR i.key = ?
                  )
              ORDER BY i.dateModified DESC
              LIMIT ?
          """, (LIBRARY_ID, pattern, pattern, pattern, query, limit)).fetchall()
  
          result = []
          for row in rows:
              item = {
                  "itemID": row["itemID"],
                  "key": row["key"],
                  "title": row["title"] or "(无标题)",
              }
              att = conn.execute("""
                  SELECT ia.itemID AS attID, ia.path
                  FROM itemAttachments ia
                  WHERE ia.parentItemID = ?
                      AND ia.contentType = 'application/pdf'
                  LIMIT 1
              """, (row["itemID"],)).fetchone()
              if att:
                  item["pdf_attachment_id"] = att["attID"]
                  item["pdf_path"] = att["path"]
              result.append(item)
  
          return result
      finally:
          conn.close()

• annota/server.py:45-58 (registration)

  FastMCP server instantiation where the tool is implicitly registered via the @mcp.tool() decorator on line 188.

  # ── MCP Server ──────────────────────────────────────────────────
  mcp = FastMCP(
      name="annota",
      instructions=(
          "Annota: 读取本地 Zotero 库中的 PDF 文件，提取带坐标的文本，"
          "并精准回写高亮批注和笔记。所有坐标均为 PDF user space（原点左下角）。\n\n"
          "【大 PDF 两阶段工作流】处理超过 10 页的论文时，请遵循：\n"
          "  Phase 1 — 理解：先用 get_pdf_text_bulk 批量提取纯文本（无坐标），让 LLM 理解全文内容，"
          "确定哪些页面的哪些句子需要标注。\n"
          "  Phase 2 — 定位：只对目标页调用 get_pdf_layout_text 获取精确坐标，"
          "然后用 create_pdf_annotation 写入标注。\n"
          "  切勿对每一页都调用 get_pdf_layout_text，这会导致上下文溢出。"
      ),
  )

• annota/server.py:10-10 (schema)

  Tool description in the module docstring: 'search_zotero_items: 按标题/作者/key 搜索条目'

  - search_zotero_items:   按标题/作者/key 搜索条目

• annota/zotero_db.py:39-74 (helper)

  The _connect() helper function used by search_items to establish a read-only database connection via copying the Zotero SQLite database.

  def _connect(readonly: bool = False) -> sqlite3.Connection:
      """创建数据库连接。
  
      Zotero 使用 EXCLUSIVE 锁模式，运行时外部无法直接读写。
      解决方案：
      - 只读操作：复制数据库到临时文件再读取（毫秒级，安全并发）
      - 写操作：直接连接 + 重试机制（等 Zotero 释放锁的间隙写入）
      """
      if readonly:
          return _connect_readonly_copy()
  
      conn = sqlite3.connect(
          f"file:{ZOTERO_DB_PATH}",
          uri=True,
          timeout=DB_TIMEOUT,
      )
      conn.row_factory = sqlite3.Row
      conn.execute(f"PRAGMA busy_timeout={DB_TIMEOUT * 1000}")
      conn.execute("PRAGMA journal_mode=WAL")
      return conn
  
  
  def _connect_readonly_copy() -> sqlite3.Connection:
      """复制数据库文件后连接，避免与 Zotero 的锁冲突。
  
      Zotero 在运行时持有排他锁，即使 WAL 模式也无法并发读。
      复制 .sqlite 文件（通常 <100MB，耗时 <100ms）到临时目录后读取。
      """
      import shutil
      import tempfile
  
      tmp_dir = Path(tempfile.gettempdir()) / "annota"
      tmp_dir.mkdir(exist_ok=True)
      tmp_db = tmp_dir / "zotero_readonly.sqlite"
  
      shutil.copy2(ZOTERO_DB_PATH, tmp_db)