This repository has been archived on 2026-07-19. You can view files and clone it. You cannot open issues or pull requests or push a commit.
Files
shira-hermes/mcp_server/tools/document_tools.py
T
chaim 90c60f45e6 feat: 9 new tools — document creation, folder management, CRM updates
Adds the Python side of the SmartAssistant 2.9 release. Each tool maps to
either a new EspoCRM action or a generic CRM endpoint.

Document creation:
- create_document: free-form RTL DOCX with markdown body; calls
  SmartAssistant/action/createDocument.

Folder management (all sandboxed to the current case's root):
- write_document_to_folder: raw upload to a path inside the case folder.
- create_subfolder: mkdir with recursive parent creation.
- rename_item / move_item: rename/move files and folders.
- mark_document_for_deletion: soft-delete — moves the file to
  "מסמכים למחיקה/" with a "למחיקה - " prefix. Shira is explicitly told
  she has NO physical-delete permission.

CRM gaps:
- update_case_fields: allow-listed PUT on Case (judge, court, caseType,
  practiceArea, opposingParty, status, priority, etc.).
- search_entities: generic GET with EspoCRM where[] params across
  Case/Contact/Account/Document/Task. Wildcard "*" maps to LIKE.
- draft_email: POST Email with status="Draft", linked to the current case.

prompt_builder additions:
- FREE DOCUMENT CREATION: don't say "I can't" — call create_document.
- FOLDER MANAGEMENT: full read/write inside the case folder only.
- FILE DELETION: always mark_document_for_deletion; never claim a physical
  delete.

Pairs with SmartAssistant extension release that adds FreeDocumentGenerator,
CaseFolderManager, and 6 controller actions.

Refs Task Master #4

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-14 13:03:59 +00:00

539 lines
25 KiB
Python

"""Document tools: list, read, read_multiple, batch_rename, generate_from_template, create_document, folder management."""
from __future__ import annotations
import base64
import json
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from mcp_server.espocrm_client import EspoCrmClient
from mcp_server.tools._helpers import ok, fail
def register_document_tools(
tools: dict,
crm: "EspoCrmClient",
case_id: str | None,
context: dict,
):
"""Register document-related tools."""
async def list_documents(refresh: bool = False) -> str:
docs = context.get("documents", {})
if refresh and case_id:
try:
docs = await crm.post("SmartAssistant/action/listDocuments", {"caseId": case_id})
context["documents"] = docs
except Exception as e:
return fail(f"שגיאה ברענון רשימת מסמכים: {e}")
# Handle legacy shape where documents is a flat list of files
if isinstance(docs, list):
if not docs:
return ok("לא נמצאו מסמכים בתיק.")
lines = [f"נמצאו {len(docs)} מסמכים:"]
for f in docs:
if isinstance(f, dict):
lines.append(f'📄 {f.get("name", "?")}{f.get("path", "?")}')
else:
lines.append(f'📄 {f}')
return ok("\n".join(lines))
if not docs.get("totalFiles"):
parts = ["לא נמצאו מסמכים בתיק."]
resolved = docs.get("resolvedPath")
source = docs.get("pathSource")
exists = docs.get("folderExists")
diag = docs.get("diagnostic")
if resolved:
parts.append(f"נתיב מחושב: '{resolved}' (מקור: {source or 'unknown'}, קיים: {exists}).")
else:
parts.append("לא ניתן לחשב נתיב תיקייה לתיק.")
if diag:
parts.append(f"סיבה: {diag}")
parts.append("הצעה: find_case_folder(query=<שם איש קשר או מספר תיק>) כדי לאתר את התיקייה, ואז set_case_folder_path(path=...) לשמור.")
return ok(" ".join(parts))
return ok(json.dumps(docs, ensure_ascii=False))
async def browse_folder(path: str) -> str:
try:
result = await crm.post("SmartAssistant/action/browseFolder", {"path": path})
entries = result.get("entries", [])
if not entries:
return ok(f'תיקייה "{path}" ריקה או לא נמצאה.')
lines = [f'תוכן "{path}" ({len(entries)} פריטים):']
for e in entries:
kind = "📁" if e.get("type") == "folder" else "📄"
size = f' ({e["size"]} bytes)' if e.get("size") else ""
lines.append(f'{kind} {e.get("name")}{e.get("path")}{size}')
return ok("\n".join(lines))
except Exception as e:
return fail(f"שגיאה בסריקת תיקייה: {e}")
async def find_case_folder(query: str, limit: int = 20) -> str:
try:
result = await crm.post("SmartAssistant/action/findCaseFolder", {"query": query, "limit": limit})
matches = result.get("matches", [])
if not matches:
return ok(f'לא נמצאו תיקיות התואמות "{query}" ב-root של האחסון.')
lines = [f'נמצאו {len(matches)} תיקיות התואמות "{query}":']
for m in matches:
lines.append(f'{m.get("name")}{m.get("path")} (רמה {m.get("level")}, score={m.get("score")})')
return ok("\n".join(lines))
except Exception as e:
return fail(f"שגיאה בחיפוש תיקייה: {e}")
async def set_case_folder_path(path: str) -> str:
if not case_id:
return fail("צריך להיות בתוך תיק כדי להגדיר נתיב תיקייה.")
try:
result = await crm.post("SmartAssistant/action/setCaseFolderPath", {"caseId": case_id, "path": path})
if not result.get("success"):
return fail(f"לא הצלחתי לשמור נתיב: {result}")
context["documents"] = {}
return ok(f'נתיב התיקייה של התיק עודכן ל-"{path}". הפעל list_documents(refresh=true) כדי לטעון את הרשימה.')
except Exception as e:
return fail(f"שגיאה בשמירת נתיב: {e}")
async def _ocr_fallback(filePath: str, reason: str) -> str:
"""Fetch file bytes and OCR via Claude Vision."""
from api.services.ocr import extract_text_from_bytes
try:
bytes_result = await crm.post("SmartAssistant/action/getDocumentBytes", {"filePath": filePath})
except Exception as e:
return fail(f'לא ניתן לחלץ טקסט מ-"{filePath}" ({reason}). כשל ב-getDocumentBytes: {e}')
if not bytes_result.get("success") or not bytes_result.get("base64"):
err = bytes_result.get("error", "unknown")
return fail(f'לא ניתן לחלץ טקסט מ-"{filePath}" ({reason}). בייטי הקובץ: {err}')
file_name = bytes_result.get("fileName", filePath)
mime_type = bytes_result.get("mimeType", "")
try:
text = await extract_text_from_bytes(
bytes_result["base64"], mime_type, file_name
)
return ok(f'=== {file_name} (OCR via Vision) ===\n\n{text}')
except Exception as e:
return fail(f'שגיאה ב-OCR של "{file_name}": {e}')
async def read_document(filePath: str) -> str:
try:
result = await crm.post("SmartAssistant/action/readDocument", {"filePath": filePath})
if not result.get("success") or not result.get("text"):
# Fallback to OCR via Claude Vision
return await _ocr_fallback(filePath, "חילוץ טקסט רגיל נכשל")
return ok(f'=== {result["fileName"]} ({result.get("charCount", "?")} תווים) ===\n\n{result["text"]}')
except Exception as e:
return fail(f"שגיאה בקריאת מסמך: {e}")
async def read_document_ocr(filePath: str) -> str:
"""Force OCR via Claude Vision — skip regular text extraction."""
return await _ocr_fallback(filePath, "OCR מבוקש במפורש")
async def read_multiple_documents(filePaths: list[str], maxCharsPerFile: int = 50000) -> str:
try:
result = await crm.post("SmartAssistant/action/readMultipleDocuments", {
"filePaths": filePaths,
"maxCharsPerFile": maxCharsPerFile,
})
lines = []
for doc in result.get("documents", []):
lines.append(f'=== {doc["fileName"]} ({doc.get("charCount", "?")} תווים) ===')
if doc.get("error"):
lines.append(f'שגיאה: {doc["error"]}')
elif not doc.get("text"):
lines.append("לא ניתן לחלץ טקסט")
else:
lines.append(doc["text"])
lines.append("")
return ok("\n".join(lines))
except Exception as e:
return fail(f"שגיאה בקריאת מסמכים: {e}")
async def batch_rename_documents(renames: list[dict]) -> str:
try:
result = await crm.post("SmartAssistant/action/batchRename", {"renames": renames})
lines = [f'שינוי שמות: {result.get("successCount", 0)}/{result.get("totalCount", 0)} הצליחו']
for r in result.get("results", []):
if r.get("success"):
lines.append(f'"{r.get("oldName", "?")}""{r.get("newName", "?")}"')
else:
lines.append(f'"{r.get("oldName", "?")}": {r.get("error", "unknown")}')
return ok("\n".join(lines))
except Exception as e:
return fail(f"שגיאה בשינוי שמות: {e}")
async def generate_from_template(
templateId: str,
documentSubject: str = "",
customPlaceholders: dict | None = None,
) -> str:
if not case_id:
return fail("צריך להיות בתוך תיק כדי לייצר מסמך מתבנית")
try:
result = await crm.post("SmartAssistant/action/generateFromTemplate", {
"templateId": templateId,
"caseId": case_id,
"documentSubject": documentSubject or None,
"customPlaceholders": customPlaceholders or {},
})
return ok(result.get("message", "✅ מסמך נוצר בהצלחה"))
except Exception as e:
return fail(f"שגיאה ביצירת מסמך מתבנית: {e}")
async def write_document_to_folder(
relPath: str,
contentBase64: str,
) -> str:
if not case_id:
return fail("צריך להיות בתוך תיק כדי לכתוב לתיקייה")
if not relPath or not relPath.strip():
return fail("חסר נתיב יחסי לתיק (relPath)")
try:
result = await crm.post("SmartAssistant/action/writeToFolder", {
"caseId": case_id,
"relPath": relPath.strip(),
"contentBase64": contentBase64,
})
path = result.get("path")
size = result.get("size", 0)
return ok(f"✅ קובץ נכתב: {path} ({size} bytes).")
except Exception as e:
return fail(f"שגיאה בכתיבת קובץ: {e}")
async def create_subfolder(relPath: str) -> str:
if not case_id:
return fail("צריך להיות בתוך תיק כדי ליצור תיקייה")
if not relPath or not relPath.strip():
return fail("חסר נתיב לתיקייה החדשה")
try:
result = await crm.post("SmartAssistant/action/createFolder", {
"caseId": case_id,
"relPath": relPath.strip(),
})
path = result.get("path")
existed = result.get("alreadyExisted", False)
verb = "כבר קיימת" if existed else "נוצרה"
return ok(f"✅ תיקייה {verb}: {path}")
except Exception as e:
return fail(f"שגיאה ביצירת תיקייה: {e}")
async def rename_item(currentRelPath: str, newName: str) -> str:
if not case_id:
return fail("צריך להיות בתוך תיק כדי לשנות שם")
if not currentRelPath or not newName:
return fail("חסר currentRelPath או newName")
try:
result = await crm.post("SmartAssistant/action/renameItem", {
"caseId": case_id,
"currentRelPath": currentRelPath.strip(),
"newName": newName.strip(),
})
return ok(f'✅ שם שונה: "{result.get("oldPath")}""{result.get("newPath")}"')
except Exception as e:
return fail(f"שגיאה בשינוי שם: {e}")
async def move_item(sourceRelPath: str, targetRelPath: str) -> str:
if not case_id:
return fail("צריך להיות בתוך תיק כדי להעביר")
if not sourceRelPath or not targetRelPath:
return fail("חסר sourceRelPath או targetRelPath")
try:
result = await crm.post("SmartAssistant/action/moveItem", {
"caseId": case_id,
"sourceRelPath": sourceRelPath.strip(),
"targetRelPath": targetRelPath.strip(),
})
return ok(f'✅ הועבר: "{result.get("oldPath")}""{result.get("newPath")}"')
except Exception as e:
return fail(f"שגיאה בהעברה: {e}")
async def mark_document_for_deletion(fileRelPath: str) -> str:
if not case_id:
return fail("צריך להיות בתוך תיק כדי לסמן למחיקה")
if not fileRelPath:
return fail("חסר fileRelPath")
try:
result = await crm.post("SmartAssistant/action/markForDeletion", {
"caseId": case_id,
"fileRelPath": fileRelPath.strip(),
})
original = result.get("originalPath")
new = result.get("newPath")
return ok(
f'📌 המסמך סומן למחיקה (לא נמחק פיזית).\n'
f'הועבר ל: {new}\n'
f'(במקור: {original}). העו"ד יוכל למחוק ידנית מתיקיית "מסמכים למחיקה".'
)
except Exception as e:
return fail(f"שגיאה בסימון למחיקה: {e}")
async def create_document(
title: str,
body: str,
documentType: str = "letter",
recipient: str = "",
) -> str:
if not case_id:
return fail("צריך להיות בתוך תיק כדי ליצור מסמך")
if not title or not title.strip():
return fail("חסר שם למסמך (title)")
if not body or not body.strip():
return fail("חסר תוכן למסמך (body)")
try:
result = await crm.post("SmartAssistant/action/createDocument", {
"caseId": case_id,
"title": title.strip(),
"body": body,
"documentType": documentType or "letter",
"recipient": recipient.strip() if recipient else None,
})
msg = result.get("message", "✅ מסמך נוצר בהצלחה")
doc_id = result.get("documentId")
net_path = result.get("networkPath")
extra = []
if doc_id:
extra.append(f"documentId={doc_id}")
if net_path:
extra.append(f"path={net_path}")
if extra:
msg += "\n" + " | ".join(extra)
return ok(msg)
except Exception as e:
return fail(f"שגיאה ביצירת מסמך: {e}")
# Register
tools["list_documents"] = {
"description": "List documents in the case folder. Shows folder structure with file names and paths. When totalFiles is 0, returns a diagnostic explaining why (missing path, folder doesn't exist, etc.). Pass refresh=true to re-fetch from EspoCRM after the folder path changes.",
"parameters": {
"type": "object",
"properties": {
"refresh": {"type": "boolean", "description": "Re-query EspoCRM instead of using cached context (default false)"},
},
},
"handler": list_documents,
}
tools["browse_folder"] = {
"description": "List the immediate contents of an arbitrary folder path on network storage. Escape hatch when list_documents returns no results — e.g. browse_folder('30-שוחר תום') to see what's really there.",
"parameters": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "Folder path relative to the storage root (e.g. '33-שוחר תום' or 'Documents/2026')"},
},
"required": ["path"],
},
"handler": browse_folder,
}
tools["find_case_folder"] = {
"description": "Search network-storage for folders whose name contains the query (e.g. contact name or case number). Scans root + one level. Returns ranked matches. Use when list_documents shows the wrong or missing path.",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Substring to match (Hebrew or English; case-insensitive)."},
"limit": {"type": "integer", "description": "Max matches to return (default 20, max 50)"},
},
"required": ["query"],
},
"handler": find_case_folder,
}
tools["set_case_folder_path"] = {
"description": "Persist the given path as the case's networkStorageFolderPath. After this, list_documents(refresh=true) will read from the new path. Use only after find_case_folder or browse_folder confirms the path exists and matches this case.",
"parameters": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "Storage-relative folder path confirmed to exist."},
},
"required": ["path"],
},
"handler": set_case_folder_path,
}
tools["read_document"] = {
"description": "Read and extract text content from a document (PDF, DOCX, TXT, images). Requires the file path from list_documents. Automatically falls back to Claude Vision OCR if regular extraction fails (e.g. scanned PDFs, image-only docx).",
"parameters": {
"type": "object",
"properties": {
"filePath": {"type": "string", "description": "Full path of the document (from list_documents results)"},
},
"required": ["filePath"],
},
"handler": read_document,
}
tools["read_document_ocr"] = {
"description": "Force OCR extraction via Claude Vision — for images (PNG/JPG/TIFF), scanned PDFs, or any file where you want AI vision to read it. Skips regular text extraction and sends the file straight to Claude Vision.",
"parameters": {
"type": "object",
"properties": {
"filePath": {"type": "string", "description": "Full path of the document (from list_documents results)"},
},
"required": ["filePath"],
},
"handler": read_document_ocr,
}
tools["read_multiple_documents"] = {
"description": "Read and extract text from multiple documents at once.",
"parameters": {
"type": "object",
"properties": {
"filePaths": {"type": "array", "items": {"type": "string"}, "description": "Array of file paths to read"},
"maxCharsPerFile": {"type": "integer", "description": "Max chars per file (default 50000)"},
},
"required": ["filePaths"],
},
"handler": read_multiple_documents,
}
tools["batch_rename_documents"] = {
"description": "Rename multiple documents at once.",
"parameters": {
"type": "object",
"properties": {
"renames": {
"type": "array",
"items": {
"type": "object",
"properties": {
"sourcePath": {"type": "string", "description": "Current file path"},
"newName": {"type": "string", "description": "New file name (without path)"},
},
"required": ["sourcePath", "newName"],
},
"description": "Array of rename operations",
},
},
"required": ["renames"],
},
"handler": batch_rename_documents,
}
tools["generate_from_template"] = {
"description": "Generate a document from any template. Templates are listed in context under availableTemplates.",
"parameters": {
"type": "object",
"properties": {
"templateId": {"type": "string", "description": "DocumentTemplate ID (from availableTemplates in context)"},
"documentSubject": {"type": "string", "description": "Custom document name"},
"customPlaceholders": {"type": "object", "description": "Extra placeholder values"},
},
"required": ["templateId"],
},
"handler": generate_from_template,
}
tools["write_document_to_folder"] = {
"description": (
"Write a new file to a path inside the current case's network folder. "
"Operations are sandboxed to the case folder — relPath is relative to the "
"case root (e.g. 'חוזים/2026/agreement.pdf'). The content must be base64-encoded. "
"Use create_document for DOCX letters; this is for raw file uploads."
),
"parameters": {
"type": "object",
"properties": {
"relPath": {"type": "string", "description": "Path inside the case folder, including filename and extension. Subfolders are created if missing."},
"contentBase64": {"type": "string", "description": "Base64-encoded file content."},
},
"required": ["relPath", "contentBase64"],
},
"handler": write_document_to_folder,
}
tools["create_subfolder"] = {
"description": (
"Create a new subfolder inside the current case's network folder. "
"relPath is relative to the case root. Multi-level paths are supported "
"(e.g. 'חוזים/2026') and missing parents are created."
),
"parameters": {
"type": "object",
"properties": {
"relPath": {"type": "string", "description": "Folder path inside the case folder."},
},
"required": ["relPath"],
},
"handler": create_subfolder,
}
tools["rename_item"] = {
"description": (
"Rename a file or folder inside the current case's network folder. "
"Does NOT move it to another folder — use move_item for that. newName "
"must be a plain name without slashes."
),
"parameters": {
"type": "object",
"properties": {
"currentRelPath": {"type": "string", "description": "Current path of the file/folder, relative to case root."},
"newName": {"type": "string", "description": "New name (without path, without slashes)."},
},
"required": ["currentRelPath", "newName"],
},
"handler": rename_item,
}
tools["move_item"] = {
"description": (
"Move a file or folder to a different location inside the current case's "
"network folder. Both source and target must stay inside the case root. "
"Missing intermediate folders on the target side are created."
),
"parameters": {
"type": "object",
"properties": {
"sourceRelPath": {"type": "string", "description": "Current path of the file/folder."},
"targetRelPath": {"type": "string", "description": "Destination path (including the final name)."},
},
"required": ["sourceRelPath", "targetRelPath"],
},
"handler": move_item,
}
tools["mark_document_for_deletion"] = {
"description": (
"Soft-delete a document: move it to a 'מסמכים למחיקה' subfolder inside the case "
"with a 'למחיקה - ' filename prefix. Does NOT physically delete. You DO NOT have "
"permission to delete files; always call this tool when the user asks you to "
"delete a document, and explain to the user that you marked it for the lawyer to "
"delete manually."
),
"parameters": {
"type": "object",
"properties": {
"fileRelPath": {"type": "string", "description": "Path to the file inside the case folder."},
},
"required": ["fileRelPath"],
},
"handler": mark_document_for_deletion,
}
tools["create_document"] = {
"description": (
"Create a new free-form DOCX document attached to the current case. "
"Use when no suitable template exists in availableTemplates. The body accepts "
"lightweight markdown: '#'/'##'/'###' for headings, '- ' for bullets, "
"'**bold**' and '*italic*' inline. The document is created as a Hebrew RTL DOCX "
"with a David font, today's date, optional recipient line, the title centered, "
"then the body. It is saved both as an EspoCRM Document attached to the case "
"and as a file in the case's network folder."
),
"parameters": {
"type": "object",
"properties": {
"title": {"type": "string", "description": "Document title shown both in the CRM record and as the centered heading."},
"body": {"type": "string", "description": "Document body. Lightweight markdown is supported. Plain paragraphs are fine."},
"documentType": {"type": "string", "enum": ["letter", "motion", "request", "general"], "description": "Document category (default 'letter')."},
"recipient": {"type": "string", "description": "Optional recipient line shown under the date (e.g. 'המוסד לביטוח לאומי')."},
},
"required": ["title", "body"],
},
"handler": create_document,
}