feat(kb): Phase 1 — Israeli National Insurance knowledge base

Adds a searchable KB for חוק הביטוח הלאומי, תקנות הביטוח הלאומי, and חוזרי
הביטוח הלאומי. Hybrid search (pgvector cosine + tsvector/trgm) fused with
Reciprocal Rank Fusion, exposed to Shira as the `search_insurance_kb` tool.

- api/services/kb/: asyncpg pool, Voyage voyage-multilingual-2 client,
  per-kind chunker (statute-by-section / circular header-aware), ingest
  pipeline with supersession (old source → superseded_by), hybrid search.
- api/routes/admin_kb.py: POST /admin/kb/ingest (multipart), GET /admin/kb/stats.
- mcp_server/tools/legal_kb_tools.py: `search_insurance_kb` registered under
  the `legal` toolset.
- pyproject.toml: +asyncpg, +python-docx, +boto3, +python-multipart.

Infra (external): pgvector/pgvector:pg16 on the shared PG, insurance_kb DB
with hnsw + gin indexes, MinIO bucket insurance-kb, KB_DATABASE_URL in
Infisical /espocrm, VOYAGE_API_KEY/VOYAGE_MODEL/KB_DATABASE_URL on the
shira-hermes Coolify app.

Refs Task Master #2

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

.taskmaster/tasks/tasks.json

@@ -0,0 +1,39 @@
+{
+  "master": {
+    "tasks": [
+      {
+        "id": "1",
+        "title": "Discovery tools + medical_case_analysis skill",
+        "description": "Add browse_folder/find_case_folder/set_case_folder_path tools, enrich list_documents with diagnostic, update prompt rules, and add medical_case_analysis skill.",
+        "details": "",
+        "testStrategy": "",
+        "status": "in-progress",
+        "dependencies": [],
+        "priority": "medium",
+        "subtasks": [],
+        "updatedAt": "2026-04-19T06:11:18.162Z"
+      },
+      {
+        "id": "2",
+        "title": "feat(kb): Phase 1 — Israeli National Insurance KB",
+        "description": "Phase 1 of the Israeli National Insurance knowledge base. Create insurance_kb database on the shared PostgreSQL instance with pgvector and pg_trgm extensions. Schema: kb_source (law/regulation/circular + supersession) and kb_chunk (content, heading_path, section_ref, tsvector, vector(1024) with HNSW cosine index). Create MinIO bucket insurance-kb. Add KB module to shira-hermes: config, asyncpg pool, Voyage client, parse+chunk+embed pipeline, hybrid search with RRF, search_insurance_kb tool (registered in mcp_server), POST /admin/kb/ingest, GET /admin/kb/stats. Store KB_DATABASE_URL in Infisical /espocrm. Validate end-to-end by ingesting 2-3 sample circulars and issuing a semantic query via the tool.",
+        "details": "See plan in conversation on 2026-04-21. Chunking strategy differs per kind: laws/regulations split by section; circulars split header-aware ~600 tokens with 80 overlap. Supersession: soft (superseded_by FK) not delete. Dependencies to add: asyncpg, voyageai (or REST via httpx), python-docx. Voyage model: voyage-multilingual-2, dim=1024, input_type=document for ingest / query for search.",
+        "testStrategy": "Manual: upload 2-3 sample PDF circulars via POST /admin/kb/ingest, verify kb_source + kb_chunk rows, verify embedding column non-null, call search_insurance_kb with a Hebrew query and confirm top-k results cite the right section_ref.",
+        "status": "in-progress",
+        "dependencies": [],
+        "priority": "high",
+        "subtasks": [],
+        "updatedAt": "2026-04-21T00:00:00.000Z"
+      }
+    ],
+    "metadata": {
+      "version": "1.0.0",
+      "lastModified": "2026-04-21T00:00:00.000Z",
+      "taskCount": 2,
+      "completedCount": 0,
+      "tags": [
+        "master"
+      ]
+    }
+  }
+}

api/main.py

@@ -9,6 +9,7 @@ from pathlib import Path
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
+from api.routes.admin_kb import router as admin_kb_router
 from api.routes.health import router as health_router
 from api.routes.smart_assistant import router as smart_assistant_router
 
@@ -34,6 +35,7 @@ app.add_middleware(
 
 app.include_router(health_router)
 app.include_router(smart_assistant_router)
+app.include_router(admin_kb_router)
 
 
 @app.on_event("startup")

api/routes/admin_kb.py

@@ -0,0 +1,66 @@
+"""Admin endpoints for managing the Israeli National Insurance KB."""
+
+from __future__ import annotations
+
+import logging
+import os
+from typing import Literal
+
+from fastapi import APIRouter, File, Form, HTTPException, Request, UploadFile
+
+from api.services.kb import ingest as kb_ingest
+
+logger = logging.getLogger("shira.admin.kb")
+
+router = APIRouter(prefix="/admin/kb", tags=["admin-kb"])
+
+
+def _verify_admin(request: Request) -> None:
+    expected = os.environ.get("ADMIN_API_KEY") or os.environ.get("API_KEY", "")
+    if not expected:
+        raise HTTPException(status_code=503, detail="Admin API key not configured")
+    provided = request.headers.get("X-Admin-Key") or request.headers.get("X-Api-Key") or ""
+    if provided != expected:
+        raise HTTPException(status_code=401, detail="Unauthorized")
+
+
+@router.post("/ingest")
+async def ingest(
+    request: Request,
+    kind: Literal["law", "regulation", "circular"] = Form(...),
+    title: str = Form(...),
+    identifier: str | None = Form(None),
+    published_at: str | None = Form(None),
+    effective_at: str | None = Form(None),
+    source_url: str | None = Form(None),
+    original_path: str | None = Form(None),
+    file: UploadFile = File(...),
+):
+    _verify_admin(request)
+    data = await file.read()
+    if not data:
+        raise HTTPException(status_code=400, detail="Empty file")
+    try:
+        result = await kb_ingest.ingest_source(
+            kind=kind,
+            title=title,
+            identifier=identifier,
+            content=data,
+            filename=file.filename or "upload",
+            published_at=published_at,
+            effective_at=effective_at,
+            source_url=source_url,
+            original_path=original_path,
+        )
+    except kb_ingest.IngestError as e:
+        raise HTTPException(status_code=422, detail=str(e))
+    except Exception as e:
+        logger.exception("[admin.kb] ingest failed")
+        raise HTTPException(status_code=500, detail=str(e))
+    return result
+
+
+@router.get("/stats")
+async def stats(request: Request):
+    _verify_admin(request)
+    return await kb_ingest.stats()

api/services/agent_runner.py

@@ -12,6 +12,7 @@ from mcp_server.espocrm_client import EspoCrmClient
 from mcp_server.tools.crm_tools import register_crm_tools
 from mcp_server.tools.document_tools import register_document_tools
 from mcp_server.tools.legal_tools import register_legal_tools
+from mcp_server.tools.legal_kb_tools import register_legal_kb_tools
 from mcp_server.tools.signature_tools import register_signature_tools
 from api.services.skills import register_skill_tools
 from api.services.memory import register_memory_tools
@@ -81,6 +82,7 @@ class AgentRunner:
             register_document_tools(tools_registry, crm, case_id, context)
         if should_register_all or "legal" in (allowed_toolsets or []):
             register_legal_tools(tools_registry, crm, case_id, context)
+            register_legal_kb_tools(tools_registry)
         if should_register_all or "signature" in (allowed_toolsets or []):
             register_signature_tools(tools_registry, crm, case_id, user_id, context)
 

api/services/kb/__init__.py

@@ -0,0 +1 @@
+"""Israeli National Insurance knowledge base (laws, regulations, circulars)."""

api/services/kb/chunker.py

@@ -0,0 +1,164 @@
+"""Text chunking — strategies per source kind.
+
+All strategies return list[dict] with keys:
+    chunk_index, heading_path, section_ref, content, token_count
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from typing import Literal
+
+# Approx tokens per char for Hebrew/mixed text (voyage-multilingual-2 tokenizer).
+_TOKENS_PER_CHAR = 0.35
+
+# A "section" in Israeli statutes: "סעיף 12", "תקנה 5", "5א.", "12(ב)"
+_SECTION_RE = re.compile(
+    r"(?m)^\s*(?:סעיף\s+|תקנה\s+|§\s*)?(\d+[א-ת]?\.?)\s*(?:\(([א-ת\d]+)\))?\s*[\.\:]",
+)
+
+# Header in circulars — markdown-ish headings, or numbered "נושא: ..." / "מס' N. ..."
+_HEADER_RE = re.compile(
+    r"(?m)^\s*(?:#{1,4}\s+|מס['׳]\s*\d+\s*[\.\:]\s*|[A-Zא-ת][^\n]{0,80}\:\s*$)",
+)
+
+
+@dataclass
+class Chunk:
+    chunk_index: int
+    heading_path: str
+    section_ref: str
+    content: str
+    token_count: int
+
+
+def _count_tokens(text: str) -> int:
+    return max(1, int(len(text) * _TOKENS_PER_CHAR))
+
+
+def _as_dicts(chunks: list[Chunk]) -> list[dict]:
+    return [
+        {
+            "chunk_index": c.chunk_index,
+            "heading_path": c.heading_path or None,
+            "section_ref": c.section_ref or None,
+            "content": c.content,
+            "token_count": c.token_count,
+        }
+        for c in chunks
+    ]
+
+
+def chunk_statute(text: str) -> list[dict]:
+    """Law / regulation: one chunk per section. Preserves section numbering."""
+    text = text.strip()
+    if not text:
+        return []
+
+    matches = list(_SECTION_RE.finditer(text))
+    if not matches:
+        # Fallback: treat the whole thing as one chunk.
+        return _as_dicts(
+            [Chunk(0, "", "", text, _count_tokens(text))]
+        )
+
+    chunks: list[Chunk] = []
+    for i, m in enumerate(matches):
+        start = m.start()
+        end = matches[i + 1].start() if i + 1 < len(matches) else len(text)
+        content = text[start:end].strip()
+        if not content:
+            continue
+        number = m.group(1).rstrip(".")
+        sub = m.group(2)
+        section_ref = f"ס' {number}" + (f"({sub})" if sub else "")
+        chunks.append(
+            Chunk(
+                chunk_index=len(chunks),
+                heading_path=section_ref,
+                section_ref=section_ref,
+                content=content,
+                token_count=_count_tokens(content),
+            )
+        )
+    return _as_dicts(chunks)
+
+
+def chunk_circular(
+    text: str,
+    target_tokens: int = 600,
+    overlap_tokens: int = 80,
+) -> list[dict]:
+    """Circular: header-aware split, then pack paragraphs up to target size with overlap."""
+    text = text.strip()
+    if not text:
+        return []
+
+    target_chars = int(target_tokens / _TOKENS_PER_CHAR)
+    overlap_chars = int(overlap_tokens / _TOKENS_PER_CHAR)
+
+    headers = list(_HEADER_RE.finditer(text))
+    segments: list[tuple[str, str]] = []  # (heading, body)
+    if not headers:
+        segments.append(("", text))
+    else:
+        if headers[0].start() > 0:
+            segments.append(("", text[: headers[0].start()].strip()))
+        for i, h in enumerate(headers):
+            end = headers[i + 1].start() if i + 1 < len(headers) else len(text)
+            heading = h.group(0).strip().rstrip(":").lstrip("#").strip()
+            body = text[h.end() : end].strip()
+            if body:
+                segments.append((heading, body))
+
+    chunks: list[Chunk] = []
+    for heading, body in segments:
+        if len(body) <= target_chars:
+            chunks.append(
+                Chunk(
+                    chunk_index=len(chunks),
+                    heading_path=heading,
+                    section_ref=heading,
+                    content=body,
+                    token_count=_count_tokens(body),
+                )
+            )
+            continue
+
+        # Pack paragraphs
+        paragraphs = [p.strip() for p in body.split("\n\n") if p.strip()]
+        buf = ""
+        for p in paragraphs:
+            if buf and len(buf) + len(p) + 2 > target_chars:
+                chunks.append(
+                    Chunk(
+                        chunk_index=len(chunks),
+                        heading_path=heading,
+                        section_ref=heading,
+                        content=buf.strip(),
+                        token_count=_count_tokens(buf),
+                    )
+                )
+                # Overlap: keep last N chars
+                buf = (buf[-overlap_chars:] + "\n\n" + p) if overlap_chars > 0 else p
+            else:
+                buf = f"{buf}\n\n{p}" if buf else p
+        if buf.strip():
+            chunks.append(
+                Chunk(
+                    chunk_index=len(chunks),
+                    heading_path=heading,
+                    section_ref=heading,
+                    content=buf.strip(),
+                    token_count=_count_tokens(buf),
+                )
+            )
+
+    return _as_dicts(chunks)
+
+
+def chunk(text: str, kind: Literal["law", "regulation", "circular"]) -> list[dict]:
+    if kind in ("law", "regulation"):
+        return chunk_statute(text)
+    return chunk_circular(text)

api/services/kb/db.py

@@ -0,0 +1,59 @@
+"""asyncpg pool for the insurance_kb database."""
+
+from __future__ import annotations
+
+import logging
+import os
+from typing import Optional
+
+import asyncpg
+
+logger = logging.getLogger("shira.kb.db")
+
+_pool: Optional[asyncpg.Pool] = None
+
+
+async def _init_connection(conn: asyncpg.Connection) -> None:
+    await conn.set_type_codec(
+        "vector",
+        schema="public",
+        encoder=_vector_encode,
+        decoder=_vector_decode,
+        format="text",
+    )
+
+
+def _vector_encode(value) -> str:
+    if value is None:
+        return None
+    return "[" + ",".join(f"{x:.8f}" for x in value) + "]"
+
+
+def _vector_decode(value: str):
+    if value is None:
+        return None
+    return [float(x) for x in value.strip("[]").split(",")]
+
+
+async def get_pool() -> asyncpg.Pool:
+    global _pool
+    if _pool is None:
+        dsn = os.environ.get("KB_DATABASE_URL")
+        if not dsn:
+            raise RuntimeError("KB_DATABASE_URL is not set")
+        _pool = await asyncpg.create_pool(
+            dsn=dsn,
+            min_size=1,
+            max_size=5,
+            init=_init_connection,
+            command_timeout=60,
+        )
+        logger.info("[kb.db] pool created")
+    return _pool
+
+
+async def close_pool() -> None:
+    global _pool
+    if _pool is not None:
+        await _pool.close()
+        _pool = None

api/services/kb/ingest.py

@@ -0,0 +1,186 @@
+"""Ingest pipeline: bytes → parse → chunk → embed → upsert."""
+
+from __future__ import annotations
+
+import hashlib
+import io
+import logging
+from typing import Literal
+
+import fitz  # pymupdf
+
+from api.services.kb import voyage
+from api.services.kb.chunker import chunk
+from api.services.kb.db import get_pool
+
+logger = logging.getLogger("shira.kb.ingest")
+
+
+class IngestError(RuntimeError):
+    pass
+
+
+def _parse_pdf(data: bytes) -> str:
+    parts: list[str] = []
+    with fitz.open(stream=data, filetype="pdf") as doc:
+        for page in doc:
+            parts.append(page.get_text("text"))
+    return "\n\n".join(parts).strip()
+
+
+def _parse_docx(data: bytes) -> str:
+    from docx import Document  # local import — keeps startup light
+    doc = Document(io.BytesIO(data))
+    return "\n\n".join(p.text for p in doc.paragraphs if p.text.strip())
+
+
+def _parse_text(data: bytes) -> str:
+    return data.decode("utf-8", errors="replace").strip()
+
+
+def parse(data: bytes, filename: str) -> str:
+    name = (filename or "").lower()
+    if name.endswith(".pdf"):
+        return _parse_pdf(data)
+    if name.endswith(".docx"):
+        return _parse_docx(data)
+    if name.endswith((".txt", ".md")):
+        return _parse_text(data)
+    # Fallback: try PDF then text
+    try:
+        return _parse_pdf(data)
+    except Exception:
+        return _parse_text(data)
+
+
+async def ingest_source(
+    *,
+    kind: Literal["law", "regulation", "circular"],
+    title: str,
+    identifier: str | None,
+    content: bytes,
+    filename: str,
+    published_at: str | None = None,
+    effective_at: str | None = None,
+    source_url: str | None = None,
+    original_path: str | None = None,
+) -> dict:
+    """Parse, chunk, embed, and upsert a source document.
+
+    Returns: {source_id, chunks_created, was_updated, checksum}
+    """
+    text = parse(content, filename)
+    if not text:
+        raise IngestError(f"no text extracted from {filename}")
+
+    checksum = hashlib.sha256(text.encode("utf-8")).hexdigest()
+    chunks = chunk(text, kind)
+    if not chunks:
+        raise IngestError("chunker produced zero chunks")
+
+    logger.info(
+        "[kb.ingest] kind=%s title=%r len=%d chunks=%d checksum=%s",
+        kind, title[:60], len(text), len(chunks), checksum[:12],
+    )
+
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        async with conn.transaction():
+            existing = await conn.fetchrow(
+                """
+                SELECT id, checksum FROM kb_source
+                WHERE kind = $1 AND COALESCE(identifier,'') = COALESCE($2,'')
+                ORDER BY created_at DESC LIMIT 1
+                """,
+                kind, identifier,
+            )
+            was_updated = False
+
+            if existing and existing["checksum"] == checksum:
+                logger.info("[kb.ingest] skip — identical checksum, source_id=%s", existing["id"])
+                return {
+                    "source_id": existing["id"],
+                    "chunks_created": 0,
+                    "was_updated": False,
+                    "checksum": checksum,
+                }
+
+            # Upsert source (new row per checksum — supersession is explicit).
+            row = await conn.fetchrow(
+                """
+                INSERT INTO kb_source
+                    (kind, title, identifier, published_at, effective_at,
+                     source_url, original_path, checksum)
+                VALUES ($1,$2,$3,$4::date,$5::date,$6,$7,$8)
+                RETURNING id
+                """,
+                kind, title, identifier, published_at, effective_at,
+                source_url, original_path, checksum,
+            )
+            source_id = row["id"]
+
+            # If a previous source exists, mark it superseded.
+            if existing:
+                await conn.execute(
+                    "UPDATE kb_source SET superseded_by = $1 WHERE id = $2",
+                    source_id, existing["id"],
+                )
+                was_updated = True
+
+            # Embed in batches via Voyage, then insert chunks.
+            texts = [c["content"] for c in chunks]
+            vectors = await voyage.embed(texts, input_type="document")
+
+            rows = [
+                (
+                    source_id,
+                    c["chunk_index"],
+                    c["heading_path"],
+                    c["section_ref"],
+                    c["content"],
+                    vectors[i],
+                    c["token_count"],
+                )
+                for i, c in enumerate(chunks)
+            ]
+            await conn.executemany(
+                """
+                INSERT INTO kb_chunk
+                    (source_id, chunk_index, heading_path, section_ref,
+                     content, embedding, token_count)
+                VALUES ($1,$2,$3,$4,$5,$6,$7)
+                """,
+                rows,
+            )
+
+    return {
+        "source_id": source_id,
+        "chunks_created": len(chunks),
+        "was_updated": was_updated,
+        "checksum": checksum,
+    }
+
+
+async def stats() -> dict:
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        row = await conn.fetchrow(
+            """
+            SELECT
+              (SELECT COUNT(*) FROM kb_source) AS total_sources,
+              (SELECT COUNT(*) FROM kb_source WHERE superseded_by IS NULL) AS active_sources,
+              (SELECT COUNT(*) FROM kb_chunk) AS total_chunks,
+              (SELECT COUNT(*) FROM kb_chunk WHERE embedding IS NOT NULL) AS embedded_chunks
+            """
+        )
+        by_kind = await conn.fetch(
+            """
+            SELECT kind, COUNT(*) AS n
+            FROM kb_source WHERE superseded_by IS NULL
+            GROUP BY kind ORDER BY kind
+            """
+        )
+    return {
+        **dict(row),
+        "active_by_kind": {r["kind"]: r["n"] for r in by_kind},
+    }

api/services/kb/search.py

@@ -0,0 +1,83 @@
+"""Hybrid search: vector + full-text, fused with Reciprocal Rank Fusion."""
+
+from __future__ import annotations
+
+import logging
+from typing import Literal
+
+from api.services.kb import voyage
+from api.services.kb.db import get_pool
+
+logger = logging.getLogger("shira.kb.search")
+
+_RRF_K = 60  # standard RRF constant
+_CANDIDATES_PER_SIDE = 30
+
+
+async def search(
+    query: str,
+    kind: Literal["law", "regulation", "circular", "any"] = "any",
+    top_k: int = 8,
+) -> list[dict]:
+    query = (query or "").strip()
+    if not query:
+        return []
+
+    [vec] = await voyage.embed([query], input_type="query")
+
+    kind_filter = ""
+    params: list = [vec, query, _CANDIDATES_PER_SIDE]
+    if kind != "any":
+        kind_filter = "AND s.kind = $4"
+        params.append(kind)
+
+    # Two CTEs — vector rank and lexical rank — fused via RRF.
+    sql = f"""
+    WITH vec AS (
+      SELECT c.id, ROW_NUMBER() OVER (ORDER BY c.embedding <=> $1::vector) AS rank
+      FROM kb_chunk c
+      JOIN kb_source s ON s.id = c.source_id
+      WHERE c.embedding IS NOT NULL
+        AND s.superseded_by IS NULL
+        {kind_filter}
+      ORDER BY c.embedding <=> $1::vector
+      LIMIT $3
+    ),
+    lex AS (
+      SELECT c.id, ROW_NUMBER() OVER (ORDER BY ts_rank_cd(c.content_tsv, q) DESC) AS rank
+      FROM kb_chunk c
+      JOIN kb_source s ON s.id = c.source_id,
+           websearch_to_tsquery('simple', $2) AS q
+      WHERE c.content_tsv @@ q
+        AND s.superseded_by IS NULL
+        {kind_filter}
+      ORDER BY ts_rank_cd(c.content_tsv, q) DESC
+      LIMIT $3
+    ),
+    fused AS (
+      SELECT id, SUM(score) AS score FROM (
+        SELECT id, 1.0 / ({_RRF_K} + rank) AS score FROM vec
+        UNION ALL
+        SELECT id, 1.0 / ({_RRF_K} + rank) AS score FROM lex
+      ) u GROUP BY id
+    )
+    SELECT
+      s.kind, s.title, s.identifier, s.source_url,
+      c.heading_path, c.section_ref, c.content,
+      f.score
+    FROM fused f
+    JOIN kb_chunk c ON c.id = f.id
+    JOIN kb_source s ON s.id = c.source_id
+    ORDER BY f.score DESC
+    LIMIT {int(top_k)}
+    """
+
+    pool = await get_pool()
+    async with pool.acquire() as conn:
+        rows = await conn.fetch(sql, *params)
+
+    logger.info(
+        "[kb.search] query=%r kind=%s hits=%d",
+        query[:80], kind, len(rows),
+    )
+    return [dict(r) for r in rows]

api/services/kb/voyage.py

@@ -0,0 +1,55 @@
+"""Voyage embeddings client — voyage-multilingual-2, 1024-dim."""
+
+from __future__ import annotations
+
+import logging
+import os
+from typing import Literal
+
+import httpx
+
+logger = logging.getLogger("shira.kb.voyage")
+
+_VOYAGE_URL = "https://api.voyageai.com/v1/embeddings"
+_MAX_BATCH = 128
+
+
+class VoyageError(RuntimeError):
+    pass
+
+
+async def embed(
+    texts: list[str],
+    input_type: Literal["document", "query"] = "document",
+    model: str | None = None,
+) -> list[list[float]]:
+    if not texts:
+        return []
+
+    api_key = os.environ.get("VOYAGE_API_KEY")
+    if not api_key:
+        raise VoyageError("VOYAGE_API_KEY is not set")
+
+    model = model or os.environ.get("VOYAGE_MODEL", "voyage-multilingual-2")
+
+    out: list[list[float]] = []
+    async with httpx.AsyncClient(timeout=60) as http:
+        for start in range(0, len(texts), _MAX_BATCH):
+            batch = texts[start : start + _MAX_BATCH]
+            resp = await http.post(
+                _VOYAGE_URL,
+                headers={"Authorization": f"Bearer {api_key}"},
+                json={"model": model, "input": batch, "input_type": input_type},
+            )
+            if resp.status_code != 200:
+                raise VoyageError(f"Voyage API {resp.status_code}: {resp.text[:200]}")
+            data = resp.json()
+            for item in sorted(data["data"], key=lambda d: d["index"]):
+                out.append(item["embedding"])
+            logger.info(
+                "[kb.voyage] embedded batch=%d/%d type=%s",
+                min(start + _MAX_BATCH, len(texts)),
+                len(texts),
+                input_type,
+            )
+    return out

mcp_server/tools/legal_kb_tools.py

@@ -0,0 +1,86 @@
+"""Israeli National Insurance knowledge base tools (search)."""
+
+from __future__ import annotations
+
+import logging
+
+from api.services.kb import search as kb_search
+from mcp_server.tools._helpers import fail
+
+logger = logging.getLogger("shira.tools.legal_kb")
+
+_KIND_HEBREW = {
+    "law": "חוק",
+    "regulation": "תקנה",
+    "circular": "חוזר",
+}
+
+
+def _format_hits(hits: list[dict]) -> str:
+    if not hits:
+        return "לא נמצאו תוצאות רלוונטיות בבסיס הידע."
+
+    lines = [f"נמצאו {len(hits)} קטעים רלוונטיים:"]
+    for i, h in enumerate(hits, 1):
+        kind_he = _KIND_HEBREW.get(h["kind"], h["kind"])
+        ident = f" {h['identifier']}" if h.get("identifier") else ""
+        ref = h.get("section_ref") or h.get("heading_path") or ""
+        header = f"{i}. [{kind_he}{ident}] {h['title']}"
+        if ref:
+            header += f" — {ref}"
+        lines.append(header)
+        content = (h.get("content") or "").strip()
+        if len(content) > 900:
+            content = content[:900] + "…"
+        lines.append(content)
+        if h.get("source_url"):
+            lines.append(f"מקור: {h['source_url']}")
+        lines.append("")
+    return "\n".join(lines).rstrip()
+
+
+def register_legal_kb_tools(tools: dict) -> None:
+    async def search_insurance_kb(
+        query: str,
+        kind: str = "any",
+        top_k: int = 8,
+    ) -> str:
+        try:
+            if kind not in ("law", "regulation", "circular", "any"):
+                kind = "any"
+            top_k = max(1, min(int(top_k or 8), 15))
+            hits = await kb_search.search(query=query, kind=kind, top_k=top_k)
+            return _format_hits(hits)
+        except Exception as e:
+            logger.exception("[search_insurance_kb] failed")
+            return fail(f"שגיאה בחיפוש בבסיס הידע: {e}")
+
+    tools["search_insurance_kb"] = {
+        "description": (
+            "חיפוש בבסיס הידע של הביטוח הלאומי — חוק הביטוח הלאומי, תקנות הביטוח "
+            "הלאומי וחוזרי הביטוח הלאומי. מחזיר קטעים רלוונטיים עם ציטוט מדויק "
+            "(סעיף/תקנה/מס' חוזר). יש להשתמש בכלי לפני טענה משפטית מבוססת-מקור, "
+            "ולצטט את ה-section_ref המדויק בתשובה. החיפוש משלב דמיון סמנטי וחיפוש "
+            "מילולי (hybrid)."
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "query": {
+                    "type": "string",
+                    "description": "שאילתת חיפוש בעברית — ניסוח חופשי או מונחים מדויקים.",
+                },
+                "kind": {
+                    "type": "string",
+                    "enum": ["law", "regulation", "circular", "any"],
+                    "description": "סינון לפי סוג מקור: law=חוק, regulation=תקנה, circular=חוזר, any=הכל (ברירת מחדל).",
+                },
+                "top_k": {
+                    "type": "integer",
+                    "description": "מספר קטעים להחזיר (1–15, ברירת מחדל 8).",
+                },
+            },
+            "required": ["query"],
+        },
+        "handler": search_insurance_kb,
+    }

pyproject.toml

@@ -11,6 +11,10 @@ dependencies = [
     "openai>=1.50.0",
     "pyyaml>=6.0",
     "pymupdf>=1.24.0",
+    "asyncpg>=0.30.0",
+    "python-docx>=1.1.2",
+    "boto3>=1.35.0",
+    "python-multipart>=0.0.9",
 ]
 
 [project.optional-dependencies]