ezer-mishpati/legal-ai
1
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

Add case_precedents: attached legal support for the compose phase

Browse Source 
New self-contained table + MCP tools + FastAPI endpoints for letting
the chair attach external case-law quotes (quote + citation מראה מקום,
optional chair note, optional archived PDF) to either a specific
threshold_claim / issue or the case as a whole.

Data model
  - case_precedents (SCHEMA_V5_SQL) — case_id, section_id NULL/
    "threshold_N"/"issue_N", quote, citation (free-text), chair_note,
    pdf_document_id FK to documents, denormalized practice_area for
    cross-case library filtering.
  - Deliberately NOT linked to the existing case_law table — that one
    has UNIQUE(case_number) which would force parsing the free-text
    citation into a structured key. A backfill pass into case_law is
    a later follow-up once the UI stabilizes.
  - db.py gains 4 helpers: create_case_precedent, list_case_precedents,
    delete_case_precedent, search_precedent_library. The last uses
    DISTINCT ON (citation) for the cross-case typeahead so each
    precedent appears once even if reused across many cases.

MCP tools (legal_mcp/tools/precedents.py)
  - precedent_attach, precedent_list, precedent_remove,
    precedent_search_library — registered in server.py.

FastAPI (web/app.py)
  - POST /api/cases/{n}/precedents — create, with PrecedentCreateRequest
  - POST /api/cases/{n}/precedents/upload-pdf — one-shot PDF upload to
    a dedicated documents/precedents/ subdirectory, creates a
    documents row with doc_type="precedent_archive" and no text
    extraction (archive only)
  - GET /api/cases/{n}/precedents — list
  - DELETE /api/precedents/{id} — uses path param since precedent_id
    is a UUID (slash-safe, unlike case numbers)
  - GET /api/precedents/search?q=...&practice_area=... — library
    typeahead

Block-writer integration into _build_precedents_context is a deferred
follow-up — Phase 1 surfaces the feature in the compose UI only.

Plan: ~/.claude/plans/woolly-cooking-graham.md

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Chaim
2026-04-11 19:16:48 +00:00
parent 8989ad9a9b
commit e2088a4f60
4 changed files with 392 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
mcp-server/src/legal_mcp/server.py
View File

@@ -45,7 +45,9 @@ mcp = FastMCP(
# ── Import and register tools ───────────────────────────────────────
from legal_mcp.tools import cases, documents, search, drafting, workflow # noqa: E402
from legal_mcp.tools import ( # noqa: E402
cases, documents, search, drafting, workflow, precedents,
)
# Case management
@@ -108,6 +110,42 @@ async def case_delete(case_number: str, remove_files: bool = False) -> str:
return await cases.case_delete(case_number, remove_files)
# Precedent attachments (user-supplied legal support for the compose phase)
@mcp.tool()
async def precedent_attach(
case_number: str,
quote: str,
citation: str,
section_id: str = "",
chair_note: str = "",
pdf_document_id: str = "",
) -> str:
"""צירוף פסיקה תומכת לתיק. section_id ריק = כללי לתיק; אחרת threshold_1/issue_3."""
return await precedents.precedent_attach(
case_number, quote, citation, section_id, chair_note, pdf_document_id,
)
@mcp.tool()
async def precedent_list(case_number: str) -> str:
"""רשימת כל הפסיקות שצורפו לתיק."""
return await precedents.precedent_list(case_number)
@mcp.tool()
async def precedent_remove(precedent_id: str) -> str:
"""הסרת פסיקה מצורפת."""
return await precedents.precedent_remove(precedent_id)
@mcp.tool()
async def precedent_search_library(
query: str, practice_area: str = "", limit: int = 10,
) -> str:
"""חיפוש בספרייה הרוחבית של ציטוטים שנצברו בין תיקים."""
return await precedents.precedent_search_library(query, practice_area, limit)
# Documents
@mcp.tool()
async def document_upload(

144
mcp-server/src/legal_mcp/services/db.py
View File

@@ -270,6 +270,40 @@ UPDATE document_chunks dc
WHERE dc.document_id = d.id AND dc.practice_area IS NULL;
"""
# ── Phase 5: case_precedents (user-attached legal quotes) ──────────
SCHEMA_V5_SQL = """
-- ═══════════════════════════════════════════════════════════════════
-- case_precedents: legal support the chair attaches to a case / section
-- during the compose phase. Self-contained — quote + citation are
-- stored inline, with an optional FK to an archived PDF in documents.
-- Not linked to case_law (which has UNIQUE(case_number)) to keep the
-- citation as free-text. A backfill pass into case_law is a future
-- follow-up once the UI stabilizes.
-- ═══════════════════════════════════════════════════════════════════
CREATE TABLE IF NOT EXISTS case_precedents (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
case_id UUID NOT NULL REFERENCES cases(id) ON DELETE CASCADE,
section_id TEXT, -- NULL = case-level
-- else "threshold_1" / "issue_3"
quote TEXT NOT NULL,
citation TEXT NOT NULL, -- free-text "מראה מקום"
chair_note TEXT DEFAULT '',
pdf_document_id UUID REFERENCES documents(id) ON DELETE SET NULL,
practice_area TEXT, -- denormalized from cases
created_at TIMESTAMPTZ DEFAULT now(),
updated_at TIMESTAMPTZ DEFAULT now()
);
CREATE INDEX IF NOT EXISTS idx_case_precedents_case
ON case_precedents(case_id, section_id);
CREATE INDEX IF NOT EXISTS idx_case_precedents_library
ON case_precedents(citation);
CREATE INDEX IF NOT EXISTS idx_case_precedents_area
ON case_precedents(practice_area);
"""
# ── Phase 2: Decision + Knowledge + RAG layers ────────────────────
SCHEMA_V2_SQL = """
@@ -459,7 +493,8 @@ async def init_schema() -> None:
await conn.execute(SCHEMA_V2_SQL)
await conn.execute(SCHEMA_V3_SQL)
await conn.execute(SCHEMA_V4_SQL)
logger.info("Database schema initialized (v1 + v2 + v3 + v4)")
await conn.execute(SCHEMA_V5_SQL)
logger.info("Database schema initialized (v1 + v2 + v3 + v4 + v5)")
# ── Case CRUD ───────────────────────────────────────────────────────
@@ -680,6 +715,113 @@ def _row_to_doc(row: asyncpg.Record) -> dict:
return d
# ── case_precedents CRUD ───────────────────────────────────────────
def _row_to_precedent(row: asyncpg.Record) -> dict:
d = dict(row)
for k in ("id", "case_id"):
if d.get(k) is not None:
d[k] = str(d[k])
if d.get("pdf_document_id") is not None:
d["pdf_document_id"] = str(d["pdf_document_id"])
for ts in ("created_at", "updated_at"):
if d.get(ts) is not None:
d[ts] = d[ts].isoformat()
return d
async def create_case_precedent(
case_id: UUID,
quote: str,
citation: str,
section_id: str | None = None,
chair_note: str = "",
pdf_document_id: UUID | None = None,
practice_area: str | None = None,
) -> dict:
"""Insert a new attached precedent. practice_area is inherited from
the parent case when not explicitly supplied, so the cross-case
library search can filter without a JOIN."""
pool = await get_pool()
async with pool.acquire() as conn:
if practice_area is None:
row = await conn.fetchrow(
"SELECT practice_area FROM cases WHERE id = $1", case_id
)
practice_area = row["practice_area"] if row else None
inserted = await conn.fetchrow(
"""INSERT INTO case_precedents
(case_id, section_id, quote, citation, chair_note,
pdf_document_id, practice_area)
VALUES ($1, $2, $3, $4, $5, $6, $7)
RETURNING *""",
case_id, section_id, quote, citation, chair_note,
pdf_document_id, practice_area,
)
return _row_to_precedent(inserted)
async def list_case_precedents(case_id: UUID) -> list[dict]:
pool = await get_pool()
async with pool.acquire() as conn:
rows = await conn.fetch(
"SELECT * FROM case_precedents WHERE case_id = $1 "
"ORDER BY section_id NULLS FIRST, created_at",
case_id,
)
return [_row_to_precedent(r) for r in rows]
async def delete_case_precedent(precedent_id: UUID) -> bool:
pool = await get_pool()
async with pool.acquire() as conn:
result = await conn.execute(
"DELETE FROM case_precedents WHERE id = $1", precedent_id
)
return result.endswith(" 1")
async def search_precedent_library(
query: str, practice_area: str = "", limit: int = 10,
) -> list[dict]:
"""Cross-case typeahead for the citation field. Returns one row per
distinct citation so the user sees each precedent once even if they
previously attached it to multiple cases/sections. No embeddings —
simple ILIKE is fine at this scale."""
pool = await get_pool()
pattern = f"%{query}%"
async with pool.acquire() as conn:
if practice_area:
rows = await conn.fetch(
"""SELECT DISTINCT ON (citation)
id, citation, quote, chair_note, practice_area, created_at
FROM case_precedents
WHERE practice_area = $1
AND (citation ILIKE $2 OR quote ILIKE $2)
ORDER BY citation, created_at DESC
LIMIT $3""",
practice_area, pattern, limit,
)
else:
rows = await conn.fetch(
"""SELECT DISTINCT ON (citation)
id, citation, quote, chair_note, practice_area, created_at
FROM case_precedents
WHERE citation ILIKE $1 OR quote ILIKE $1
ORDER BY citation, created_at DESC
LIMIT $2""",
pattern, limit,
)
out = []
for r in rows:
d = dict(r)
d["id"] = str(d["id"])
if d.get("created_at"):
d["created_at"] = d["created_at"].isoformat()
out.append(d)
return out
# ── Claims ─────────────────────────────────────────────────────────
async def store_claims(case_id: UUID, claims: list[dict], source_document: str = "") -> int:

95
mcp-server/src/legal_mcp/tools/precedents.py Normal file
View File

@@ -0,0 +1,95 @@
"""MCP tools for attached legal precedents (user-supplied case-law quotes).
These complement the existing `case_law` table (which is populated from
structured sources and is what the block-writer RAG searches) by storing
free-text citations the chair attaches during the compose phase.
"""
from __future__ import annotations
import json
from pathlib import Path
from uuid import UUID
from legal_mcp.services import db
async def precedent_attach(
case_number: str,
quote: str,
citation: str,
section_id: str = "",
chair_note: str = "",
pdf_document_id: str = "",
) -> str:
"""צירוף פסיקה תומכת לתיק ערר.
Args:
case_number: מספר תיק הערר
quote: הציטוט המדויק שיוכנס להחלטה
citation: מראה המקום (ערר 1126-08-25 ... נ' ... (נבו 9.3.2026))
section_id: מזהה הטענה/סוגיה (threshold_1, issue_3); ריק = כללי לתיק
chair_note: הערה אופציונלית — למה הציטוט תומך בעמדה
pdf_document_id: מזהה קובץ PDF מצורף (אופציונלי)
"""
case = await db.get_case_by_number(case_number)
if not case:
return json.dumps({"error": f"תיק {case_number} לא נמצא."}, ensure_ascii=False)
pdf_uuid: UUID | None = None
if pdf_document_id:
try:
pdf_uuid = UUID(pdf_document_id)
except ValueError:
return json.dumps({"error": "pdf_document_id לא תקין"}, ensure_ascii=False)
row = await db.create_case_precedent(
case_id=UUID(case["id"]),
quote=quote,
citation=citation,
section_id=section_id or None,
chair_note=chair_note,
pdf_document_id=pdf_uuid,
practice_area=case.get("practice_area"),
)
return json.dumps(row, ensure_ascii=False, indent=2)
async def precedent_list(case_number: str) -> str:
"""רשימת כל הפסיקות שצורפו לתיק, ממוינות לפי סעיף ואז לפי זמן יצירה."""
case = await db.get_case_by_number(case_number)
if not case:
return json.dumps({"error": f"תיק {case_number} לא נמצא."}, ensure_ascii=False)
rows = await db.list_case_precedents(UUID(case["id"]))
return json.dumps(rows, ensure_ascii=False, indent=2)
async def precedent_remove(precedent_id: str) -> str:
"""הסרת פסיקה מצורפת. קובץ ה-PDF (אם צורף) נשאר ב-documents לצורך audit."""
try:
pid = UUID(precedent_id)
except ValueError:
return json.dumps({"error": "precedent_id לא תקין"}, ensure_ascii=False)
ok = await db.delete_case_precedent(pid)
return json.dumps(
{"deleted": ok, "precedent_id": precedent_id}, ensure_ascii=False,
)
async def precedent_search_library(
query: str, practice_area: str = "", limit: int = 10,
) -> str:
"""חיפוש בספרייה הרוחבית — כל הפסיקות שצורפו אי-פעם בכל התיקים.
Args:
query: מחרוזת חיפוש (מתחרה מול citation ומול quote)
practice_area: אופציונלי — סינון לתחום משפטי מסוים
limit: מספר תוצאות מקסימלי
"""
if not query or len(query.strip()) < 2:
return json.dumps([], ensure_ascii=False)
rows = await db.search_precedent_library(query.strip(), practice_area, limit)
return json.dumps(rows, ensure_ascii=False, indent=2)

116
web/app.py
View File

@@ -29,7 +29,7 @@ import asyncpg
from legal_mcp import config
from legal_mcp.services import chunker, db, embeddings, extractor, processor, proofreader, research_md
from legal_mcp.tools import cases as cases_tools, search as search_tools, workflow as workflow_tools, drafting as drafting_tools
from legal_mcp.tools import cases as cases_tools, search as search_tools, workflow as workflow_tools, drafting as drafting_tools, precedents as precedents_tools
# Import integration clients (same directory)
_web_dir = Path(__file__).resolve().parent
@@ -1643,6 +1643,120 @@ async def api_research_chair_position(case_number: str, req: ChairPositionReques
raise HTTPException(500, f"שגיאה בשמירה: {e}")
# ── Precedents API — attached case-law quotes for the compose phase ──
class PrecedentCreateRequest(BaseModel):
quote: str
citation: str
section_id: str = "" # empty = case-level / general discussion
chair_note: str = ""
pdf_document_id: str = "" # UUID string, empty = no PDF
@app.post("/api/cases/{case_number}/precedents")
async def api_precedent_attach(case_number: str, req: PrecedentCreateRequest):
"""Attach a legal precedent (quote + citation) to a case, optionally
scoped to a specific threshold_claim / issue section. Cross-case
library reuse happens at the search endpoint — this one always
inserts a new row."""
if req.section_id and not re.match(r"^(threshold|issue)_\d+$", req.section_id):
raise HTTPException(400, "section_id לא תקין")
if not req.quote.strip() or not req.citation.strip():
raise HTTPException(400, "quote ו-citation חובה")
result = await precedents_tools.precedent_attach(
case_number=case_number,
quote=req.quote,
citation=req.citation,
section_id=req.section_id,
chair_note=req.chair_note,
pdf_document_id=req.pdf_document_id,
)
data = json.loads(result)
if data.get("error"):
raise HTTPException(404, data["error"])
return data
@app.post("/api/cases/{case_number}/precedents/upload-pdf")
async def api_precedent_upload_pdf(
case_number: str,
file: UploadFile = File(...),
):
"""One-shot PDF upload for a precedent attachment. Stores the file
on disk alongside other case documents and creates a `documents`
row with doc_type='precedent_archive'. Returns {document_id} so the
frontend can pass it into POST /precedents. No SSE / background
processing — archive only, no text extraction."""
case = await db.get_case_by_number(case_number)
if not case:
raise HTTPException(404, f"תיק {case_number} לא נמצא")
if not file.filename:
raise HTTPException(400, "No filename provided")
ext = Path(file.filename).suffix.lower()
if ext not in {".pdf", ".docx", ".doc"}:
raise HTTPException(400, f"סוג קובץ לא נתמך לפסיקה: {ext}")
content = await file.read()
if len(content) > MAX_FILE_SIZE:
raise HTTPException(400, f"קובץ גדול מדי. מקסימום: {MAX_FILE_SIZE // (1024*1024)}MB")
# Save under a dedicated precedents/ subdirectory so they don't mix
# with extracted originals.
case_dir = config.find_case_dir(case_number) / "documents" / "precedents"
case_dir.mkdir(parents=True, exist_ok=True)
safe_name = re.sub(r"[^\w\u0590-\u05FF\s.\-()]", "", Path(file.filename).stem).strip()
dest = case_dir / f"{safe_name or 'precedent'}{ext}"
counter = 1
while dest.exists():
dest = case_dir / f"{safe_name or 'precedent'}-{counter}{ext}"
counter += 1
dest.write_bytes(content)
case_id = UUID(case["id"])
doc = await db.create_document(
case_id=case_id,
doc_type="precedent_archive",
title=safe_name or "precedent",
file_path=str(dest),
)
return {"document_id": doc["id"], "filename": dest.name}
@app.get("/api/cases/{case_number}/precedents")
async def api_precedent_list(case_number: str):
"""List all precedents attached to a case, grouped client-side by section_id."""
result = await precedents_tools.precedent_list(case_number)
data = json.loads(result)
if isinstance(data, dict) and data.get("error"):
raise HTTPException(404, data["error"])
return data
@app.delete("/api/precedents/{precedent_id}")
async def api_precedent_delete(precedent_id: str):
"""Delete a precedent attachment. The archived PDF (if any) stays
in the documents table — orphaned references nullify via FK
ON DELETE SET NULL — so we keep the audit trail of the file."""
result = await precedents_tools.precedent_remove(precedent_id)
data = json.loads(result)
if data.get("error"):
raise HTTPException(400, data["error"])
if not data.get("deleted"):
raise HTTPException(404, "לא נמצא")
return data
@app.get("/api/precedents/search")
async def api_precedent_search(q: str, practice_area: str = "", limit: int = 10):
"""Cross-case library typeahead. Returns one row per distinct citation."""
result = await precedents_tools.precedent_search_library(q, practice_area, limit)
return json.loads(result)
# ── Exports API — drafts, versions, download, upload, mark-final ──