Release v0.2.41

feat: 8 meeting-agent tools (transcribe, CRUD, summarize, extract, index, search)

praisonai_tools/tools/meeting_tools.py

@@ -0,0 +1,700 @@
+"""Meeting Agent Tools for PraisonAI Agents.
+
+Eight agent-callable ``@tool`` functions covering the Phase 1 meeting-agent
+pipeline: transcription, meetings CRUD, summarisation, action-item extraction,
+and RAG indexing/search.
+
+Tools
+-----
+1. ``transcribe_file``       - Whisper transcription with segment timestamps.
+2. ``save_meeting``          - Persist a meeting record (returns ``meeting_id``).
+3. ``get_meeting``           - Fetch a full meeting record.
+4. ``list_meetings``         - List meetings (limit/offset).
+5. ``summarize_transcript``  - Structured summary (map-reduce for long input).
+6. ``extract_action_items``  - Structured action items.
+7. ``index_meeting``         - Chunk/embed/upsert into a vector store (idempotent).
+8. ``search_meetings``       - Ranked cross-meeting semantic search.
+
+Dependency hygiene
+-------------------
+``openai`` and ``chromadb`` are **lazy-imported inside functions**, never at
+module import time. They are optional extras::
+
+    pip install praisonai-tools[meeting]
+
+Meetings are persisted in a small SQLite database (Python stdlib) under the app
+data dir. This is a thin, swappable persistence layer: when the canonical
+PAI-MEET-CORE schema is available it can back these same CRUD wrappers.
+
+All tool outputs use deterministic serialisation (stable field order / sorted
+keys) so results are reproducible.
+
+Environment Variables
+---------------------
+    OPENAI_API_KEY      Required for transcription/summaries/embeddings.
+    OPENAI_BASE_URL     Optional custom OpenAI-compatible endpoint.
+    PRAISONAI_MEETINGS_DB   Optional path to the SQLite meetings database.
+    PRAISONAI_MEETINGS_DIR  Optional app data dir (db + vector store live here).
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import sqlite3
+import uuid
+from contextlib import closing
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+from praisonai_tools.tools.decorator import tool
+
+logger = logging.getLogger(__name__)
+
+# Chunking / embedding defaults (issue spec).
+_CHUNK_SIZE_TOKENS = 500
+_CHUNK_OVERLAP_TOKENS = 50
+_SUMMARY_CHUNK_TOKENS = 6000
+_SUMMARY_OVERLAP_TOKENS = 200
+_EMBEDDING_MODEL = "text-embedding-3-small"
+_CHAT_MODEL = "gpt-4o-mini"
+_WHISPER_MODEL = "whisper-1"
+_COLLECTION = "meetings"
+
+# Rough token->char ratio used only for deterministic, dependency-free chunking.
+_CHARS_PER_TOKEN = 4
+
+
+# ── Paths / persistence ─────────────────────────────────────────────
+
+
+def _app_dir() -> Path:
+    """Return the app data directory (created if missing)."""
+    env = os.getenv("PRAISONAI_MEETINGS_DIR")
+    base = Path(env) if env else Path.home() / ".praisonai" / "meetings"
+    base.mkdir(parents=True, exist_ok=True)
+    return base
+
+
+def _db_path() -> str:
+    env = os.getenv("PRAISONAI_MEETINGS_DB")
+    if env:
+        Path(env).parent.mkdir(parents=True, exist_ok=True)
+        return env
+    return str(_app_dir() / "meetings.db")
+
+
+def _connect() -> sqlite3.Connection:
+    """Open the meetings DB, creating the schema on first use."""
+    conn = sqlite3.connect(_db_path())
+    conn.row_factory = sqlite3.Row
+    conn.execute(
+        """
+        CREATE TABLE IF NOT EXISTS meetings (
+            meeting_id TEXT PRIMARY KEY,
+            title      TEXT NOT NULL,
+            source     TEXT,
+            metadata   TEXT,
+            created_at TEXT NOT NULL
+        )
+        """
+    )
+    return conn
+
+
+def _row_to_meeting(row: sqlite3.Row) -> Dict[str, Any]:
+    """Convert a DB row to a deterministic meeting dict (stable key order)."""
+    try:
+        metadata = json.loads(row["metadata"]) if row["metadata"] else {}
+    except (ValueError, TypeError):
+        metadata = {}
+    return {
+        "created_at": row["created_at"],
+        "meeting_id": row["meeting_id"],
+        "metadata": metadata,
+        "source": row["source"],
+        "title": row["title"],
+    }
+
+
+# ── Lazy clients ────────────────────────────────────────────────────
+
+
+def _get_openai_client():
+    """Lazily construct an OpenAI client. Raises clear errors on misconfig."""
+    api_key = os.getenv("OPENAI_API_KEY")
+    if not api_key:
+        raise ValueError(
+            "OPENAI_API_KEY is required. Set it in the environment before "
+            "calling meeting transcription/summarisation/embedding tools."
+        )
+    try:
+        from openai import OpenAI
+    except ImportError as exc:  # pragma: no cover - trivial
+        raise ImportError(
+            "openai package required. Install with: pip install praisonai-tools[meeting]"
+        ) from exc
+    base_url = os.getenv("OPENAI_BASE_URL")
+    if base_url:
+        return OpenAI(api_key=api_key, base_url=base_url)
+    return OpenAI(api_key=api_key)
+
+
+def _get_vector_collection():
+    """Lazily open (or create) the persistent Chroma collection."""
+    try:
+        import chromadb
+    except ImportError as exc:  # pragma: no cover - trivial
+        raise ImportError(
+            "chromadb package required. Install with: pip install praisonai-tools[meeting]"
+        ) from exc
+    client = chromadb.PersistentClient(path=str(_app_dir() / "chroma"))
+    return client.get_or_create_collection(_COLLECTION)
+
+
+def _embed(texts: List[str]) -> List[List[float]]:
+    """Embed a list of texts with the configured embedding model."""
+    client = _get_openai_client()
+    resp = client.embeddings.create(model=_EMBEDDING_MODEL, input=texts)
+    return [item.embedding for item in resp.data]
+
+
+# ── Chunking helpers ────────────────────────────────────────────────
+
+
+def _chunk_spans(text: str, size_tokens: int, overlap_tokens: int) -> List[tuple]:
+    """Split text into overlapping windows, returning (chunk, char_start, char_end).
+
+    Uses a deterministic character-window approximation so no tokenizer
+    dependency is needed. Character offsets refer to the pre-strip window and
+    let callers map each chunk back to source positions (e.g. segment times).
+    """
+    if not text:
+        return []
+    size = max(1, size_tokens * _CHARS_PER_TOKEN)
+    overlap = max(0, min(overlap_tokens * _CHARS_PER_TOKEN, size - 1))
+    step = size - overlap
+    spans: List[tuple] = []
+    start = 0
+    length = len(text)
+    while start < length:
+        end = min(start + size, length)
+        chunk = text[start:end].strip()
+        if chunk:
+            spans.append((chunk, start, end))
+        start += step
+    return spans
+
+
+def _chunk_text(text: str, size_tokens: int, overlap_tokens: int) -> List[str]:
+    """Split text into overlapping chunks by approximate token size.
+
+    Returns a list of non-empty chunk strings (see :func:`_chunk_spans`).
+    """
+    return [chunk for chunk, _s, _e in _chunk_spans(text, size_tokens, overlap_tokens)]
+
+
+def _segment_char_offsets(segments: List[Dict[str, Any]]) -> List[tuple]:
+    """Return (char_start, char_end, seg_start, seg_end) for concatenated segments.
+
+    The transcript indexed here is assumed to be the segment texts joined by a
+    single space, matching how ``transcribe_file`` emits ``segments``.
+    """
+    offsets: List[tuple] = []
+    pos = 0
+    for i, seg in enumerate(segments):
+        seg_text = str(seg.get("text", "")).strip()
+        if i > 0:
+            pos += 1  # the joining space
+        char_start = pos
+        pos += len(seg_text)
+        offsets.append(
+            (char_start, pos, float(seg.get("start", 0.0)), float(seg.get("end", 0.0)))
+        )
+    return offsets
+
+
+def _timestamps_for_span(
+    char_start: int, char_end: int, seg_offsets: List[tuple]
+) -> tuple:
+    """Map a chunk's char span to the (start, end) time of the segments it covers."""
+    if not seg_offsets:
+        return 0.0, 0.0
+    start_time: Optional[float] = None
+    end_time: Optional[float] = None
+    for cs, ce, s_start, s_end in seg_offsets:
+        if ce > char_start and cs < char_end:  # overlap
+            if start_time is None:
+                start_time = s_start
+            end_time = s_end
+    if start_time is None:
+        return 0.0, 0.0
+    return start_time, end_time if end_time is not None else start_time
+
+
+# ── LLM JSON helpers ────────────────────────────────────────────────
+
+
+def _chat_json(system: str, user: str) -> Dict[str, Any]:
+    """Call the chat model requesting a JSON object and parse it."""
+    client = _get_openai_client()
+    resp = client.chat.completions.create(
+        model=_CHAT_MODEL,
+        messages=[
+            {"role": "system", "content": system},
+            {"role": "user", "content": user},
+        ],
+        response_format={"type": "json_object"},
+        temperature=0,
+    )
+    content = resp.choices[0].message.content or "{}"
+    try:
+        return json.loads(content)
+    except (ValueError, TypeError):
+        return {}
+
+
+# ── 1. transcribe_file ──────────────────────────────────────────────
+
+
+@tool
+def transcribe_file(media_path: str) -> Dict[str, Any]:
+    """Transcribe an audio/video file to text with segment timestamps.
+
+    Uses OpenAI Whisper (``whisper-1``) via the audio transcriptions API and
+    returns word/segment level timing.
+
+    Args:
+        media_path: Path to the audio or video file to transcribe.
+
+    Returns:
+        Dict with ``text``, ``segments`` (start/end/text), ``duration_seconds``
+        and ``language``. On failure returns ``{"error": "..."}`` with a clear
+        message (missing file, API 403 key permissions, timeout, empty audio).
+    """
+    path = Path(media_path)
+    if not media_path or not path.exists():
+        return {"error": f"Media file not found: {media_path}"}
+
+    try:
+        client = _get_openai_client()
+    except (ValueError, ImportError) as exc:
+        return {"error": str(exc)}
+
+    def _do_transcribe():
+        with open(path, "rb") as audio_file:
+            return client.audio.transcriptions.create(
+                model=_WHISPER_MODEL,
+                file=audio_file,
+                response_format="verbose_json",
+                timestamp_granularities=["segment"],
+            )
+
+    try:
+        response = _do_transcribe()
+    except Exception as exc:  # noqa: BLE001 - map to clear messages
+        message = str(exc)
+        status = getattr(exc, "status_code", None)
+        exc_name = type(exc).__name__.lower()
+        if status == 403 or "403" in message or "permission" in message.lower():
+            return {"error": "Transcription failed: API key lacks permissions (403)."}
+        if "timeout" in exc_name or "timeout" in message.lower():
+            # Retry once, then fail with a clear message.
+            try:
+                response = _do_transcribe()
+            except Exception:  # noqa: BLE001
+                return {"error": "Transcription failed: request timed out after retry."}
+        else:
+            return {"error": f"Transcription failed: {message}"}
+
+    text = (getattr(response, "text", "") or "").strip()
+    segments_raw = getattr(response, "segments", None) or []
+    segments = [
+        {
+            "end": float(getattr(seg, "end", 0.0)),
+            "start": float(getattr(seg, "start", 0.0)),
+            "text": (getattr(seg, "text", "") or "").strip(),
+        }
+        for seg in segments_raw
+    ]
+
+    if not text and not segments:
+        return {"error": "No speech detected in recording"}
+
+    return {
+        "duration_seconds": float(getattr(response, "duration", 0.0) or 0.0),
+        "language": getattr(response, "language", "unknown") or "unknown",
+        "segments": segments,
+        "text": text,
+    }
+
+
+# ── 2-4. Meetings CRUD ──────────────────────────────────────────────
+
+
+@tool
+def save_meeting(
+    title: str,
+    source: Optional[str] = None,
+    metadata: Optional[Dict[str, Any]] = None,
+) -> Dict[str, Any]:
+    """Persist a meeting record and return its generated ``meeting_id``.
+
+    Thin wrapper over the meetings persistence layer (SQLite by default).
+
+    Args:
+        title: Human-readable meeting title.
+        source: Optional origin of the meeting (e.g. file path, URL).
+        metadata: Optional JSON-serialisable metadata dict.
+
+    Returns:
+        Dict ``{"meeting_id": "<uuid>"}`` or ``{"error": "..."}``.
+    """
+    if not title:
+        return {"error": "title is required"}
+    meeting_id = str(uuid.uuid4())
+    from datetime import datetime, timezone
+
+    created_at = datetime.now(timezone.utc).isoformat()
+    try:
+        with closing(_connect()) as conn, conn:
+            conn.execute(
+                "INSERT INTO meetings (meeting_id, title, source, metadata, created_at) "
+                "VALUES (?, ?, ?, ?, ?)",
+                (
+                    meeting_id,
+                    title,
+                    source,
+                    json.dumps(metadata or {}, sort_keys=True),
+                    created_at,
+                ),
+            )
+    except Exception as exc:  # noqa: BLE001
+        logger.error("save_meeting error: %s", exc)
+        return {"error": str(exc)}
+    return {"meeting_id": meeting_id}
+
+
+@tool
+def get_meeting(meeting_id: str) -> Dict[str, Any]:
+    """Fetch a full meeting record by id.
+
+    Args:
+        meeting_id: The meeting identifier returned by ``save_meeting``.
+
+    Returns:
+        The meeting record dict, or ``{"error": "..."}`` if not found.
+    """
+    if not meeting_id:
+        return {"error": "meeting_id is required"}
+    try:
+        with closing(_connect()) as conn:
+            row = conn.execute(
+                "SELECT * FROM meetings WHERE meeting_id = ?", (meeting_id,)
+            ).fetchone()
+    except Exception as exc:  # noqa: BLE001
+        logger.error("get_meeting error: %s", exc)
+        return {"error": str(exc)}
+    if row is None:
+        return {"error": f"Meeting not found: {meeting_id}"}
+    return _row_to_meeting(row)
+
+
+@tool
+def list_meetings(limit: int = 20, offset: int = 0) -> List[Dict[str, Any]]:
+    """List stored meetings, most recent first.
+
+    Args:
+        limit: Maximum number of meetings to return (default 20).
+        offset: Number of meetings to skip for pagination (default 0).
+
+    Returns:
+        A list of meeting record dicts (possibly empty), or a single-element
+        list ``[{"error": "..."}]`` on failure.
+    """
+    try:
+        limit = max(0, int(limit))
+        offset = max(0, int(offset))
+        with closing(_connect()) as conn:
+            rows = conn.execute(
+                "SELECT * FROM meetings ORDER BY created_at DESC, meeting_id DESC "
+                "LIMIT ? OFFSET ?",
+                (limit, offset),
+            ).fetchall()
+    except Exception as exc:  # noqa: BLE001
+        logger.error("list_meetings error: %s", exc)
+        return [{"error": str(exc)}]
+    return [_row_to_meeting(r) for r in rows]
+
+
+# ── 5. summarize_transcript ─────────────────────────────────────────
+
+
+def _summarize_chunk(transcript: str) -> Dict[str, Any]:
+    system = (
+        "You are a meeting summariser. Return a JSON object with keys: "
+        "'summary' (2-3 paragraph string), 'decisions' (array of strings), "
+        "'topics' (array of strings), and 'key_quotes' (array of objects with "
+        "'speaker' and 'quote'). Base output only on the transcript."
+    )
+    return _chat_json(system, transcript)
+
+
+def _merge_summaries(parts: List[Dict[str, Any]]) -> Dict[str, Any]:
+    merged_summary = "\n\n".join(
+        str(p.get("summary", "")).strip() for p in parts if p.get("summary")
+    )
+    decisions: List[str] = []
+    topics: List[str] = []
+    quotes: List[Dict[str, Any]] = []
+    for p in parts:
+        for d in p.get("decisions", []) or []:
+            if d not in decisions:
+                decisions.append(d)
+        for t in p.get("topics", []) or []:
+            if t not in topics:
+                topics.append(t)
+        for q in p.get("key_quotes", []) or []:
+            quotes.append(
+                {"quote": q.get("quote", ""), "speaker": q.get("speaker")}
+            )
+    return {
+        "decisions": decisions,
+        "key_quotes": quotes,
+        "summary": merged_summary,
+        "topics": topics,
+    }
+
+
+@tool
+def summarize_transcript(transcript: str) -> Dict[str, Any]:
+    """Summarise a meeting transcript into structured JSON.
+
+    For long transcripts a map-reduce strategy is used: the transcript is split
+    into overlapping chunks, each is summarised, and the partial summaries are
+    merged.
+
+    Args:
+        transcript: The full plain-text transcript.
+
+    Returns:
+        Dict with ``summary`` (str), ``decisions`` (list), ``topics`` (list) and
+        ``key_quotes`` (list of ``{speaker, quote}``). ``{"error": "..."}`` on
+        failure.
+    """
+    if not transcript or not transcript.strip():
+        return {"error": "transcript is required"}
+    try:
+        chunks = _chunk_text(
+            transcript, _SUMMARY_CHUNK_TOKENS, _SUMMARY_OVERLAP_TOKENS
+        )
+        if len(chunks) <= 1:
+            result = _summarize_chunk(transcript)
+        else:
+            partials = [_summarize_chunk(c) for c in chunks]
+            result = _merge_summaries(partials)
+    except (ValueError, ImportError) as exc:
+        return {"error": str(exc)}
+    except Exception as exc:  # noqa: BLE001
+        logger.error("summarize_transcript error: %s", exc)
+        return {"error": str(exc)}
+
+    return {
+        "decisions": result.get("decisions", []) or [],
+        "key_quotes": result.get("key_quotes", []) or [],
+        "summary": result.get("summary", "") or "",
+        "topics": result.get("topics", []) or [],
+    }
+
+
+# ── 6. extract_action_items ─────────────────────────────────────────
+
+
+@tool
+def extract_action_items(transcript: str) -> Dict[str, Any]:
+    """Extract action items from a transcript as structured JSON.
+
+    Args:
+        transcript: The full plain-text transcript.
+
+    Returns:
+        Dict ``{"action_items": [{description, owner, due_date, priority}, ...]}``
+        or ``{"error": "..."}`` on failure.
+    """
+    if not transcript or not transcript.strip():
+        return {"error": "transcript is required"}
+    system = (
+        "You extract action items from meeting transcripts. Return a JSON object "
+        "with a single key 'action_items': an array of objects each with keys "
+        "'description' (string), 'owner' (string or null), 'due_date' (ISO date "
+        "string or null), and 'priority' (one of 'low', 'medium', 'high' or null). "
+        "Only include real, actionable items."
+    )
+    try:
+        result = _chat_json(system, transcript)
+    except (ValueError, ImportError) as exc:
+        return {"error": str(exc)}
+    except Exception as exc:  # noqa: BLE001
+        logger.error("extract_action_items error: %s", exc)
+        return {"error": str(exc)}
+
+    items = result.get("action_items", []) or []
+    normalised = [
+        {
+            "description": it.get("description", ""),
+            "due_date": it.get("due_date"),
+            "owner": it.get("owner"),
+            "priority": it.get("priority"),
+        }
+        for it in items
+        if isinstance(it, dict)
+    ]
+    return {"action_items": normalised}
+
+
+# ── 7. index_meeting ────────────────────────────────────────────────
+
+
+@tool
+def index_meeting(
+    meeting_id: str,
+    transcript: Optional[str] = None,
+    segments: Optional[List[Dict[str, Any]]] = None,
+) -> Dict[str, Any]:
+    """Chunk, embed and upsert a meeting transcript into the vector store.
+
+    Idempotent: existing chunks for ``meeting_id`` are replaced only *after* the
+    new embeddings are computed, so a failed re-index never leaves the meeting
+    unsearchable.
+
+    Args:
+        meeting_id: The meeting identifier.
+        transcript: Optional transcript text. If omitted, the transcript is read
+            from the stored meeting metadata (``metadata['transcript']``).
+        segments: Optional ``transcribe_file`` segments (``start``/``end``/``text``).
+            When provided, each indexed chunk is tagged with the real
+            ``timestamp_start``/``timestamp_end`` of the segments it covers.
+            Falls back to stored ``metadata['segments']`` when omitted.
+
+    Returns:
+        Dict ``{"chunks_indexed": N}`` or ``{"error": "..."}``.
+    """
+    if not meeting_id:
+        return {"error": "meeting_id is required"}
+
+    meeting = get_meeting(meeting_id)
+    if "error" in meeting:
+        return meeting
+
+    text = transcript
+    title = meeting.get("title", "")
+    meta = meeting.get("metadata") or {}
+    if not text:
+        text = meta.get("transcript")
+    if not text or not str(text).strip():
+        return {"error": "No transcript available to index"}
+
+    if segments is None:
+        segments = meta.get("segments")
+    seg_offsets = _segment_char_offsets(segments) if segments else []
+
+    try:
+        collection = _get_vector_collection()
+
+        spans = _chunk_spans(text, _CHUNK_SIZE_TOKENS, _CHUNK_OVERLAP_TOKENS)
+        if not spans:
+            return {"chunks_indexed": 0}
+
+        chunks = [c for c, _s, _e in spans]
+        # Embed first; only replace existing chunks once embedding succeeds so a
+        # failure never destroys the previously searchable version.
+        embeddings = _embed(chunks)
+
+        ids = [f"{meeting_id}:{i}" for i in range(len(chunks))]
+        metadatas = []
+        for i, (_chunk, cs, ce) in enumerate(spans):
+            ts_start, ts_end = _timestamps_for_span(cs, ce, seg_offsets)
+            metadatas.append(
+                {
+                    "chunk_index": i,
+                    "meeting_id": meeting_id,
+                    "timestamp_end": ts_end,
+                    "timestamp_start": ts_start,
+                    "title": title,
+                }
+            )
+
+        # Delete existing chunks for this meeting so re-index is idempotent.
+        collection.delete(where={"meeting_id": meeting_id})
+        collection.add(
+            ids=ids,
+            documents=chunks,
+            embeddings=embeddings,
+            metadatas=metadatas,
+        )
+    except (ValueError, ImportError) as exc:
+        return {"error": str(exc)}
+    except Exception as exc:  # noqa: BLE001
+        logger.error("index_meeting error: %s", exc)
+        return {"error": str(exc)}
+
+    return {"chunks_indexed": len(chunks)}
+
+
+# ── 8. search_meetings ──────────────────────────────────────────────
+
+
+@tool
+def search_meetings(query: str, limit: int = 5) -> List[Dict[str, Any]]:
+    """Semantic search across indexed meetings, returning ranked snippets.
+
+    Args:
+        query: The natural-language search query.
+        limit: Maximum number of results to return (default 5).
+
+    Returns:
+        A list of ``{meeting_id, title, snippet, score, timestamp_start,
+        timestamp_end}`` dicts ranked by relevance (best first), or
+        ``[{"error": "..."}]`` on failure.
+    """
+    if not query or not query.strip():
+        return [{"error": "query is required"}]
+    try:
+        limit = max(1, int(limit))
+        collection = _get_vector_collection()
+        query_embedding = _embed([query])[0]
+        results = collection.query(
+            query_embeddings=[query_embedding], n_results=limit
+        )
+    except (ValueError, ImportError) as exc:
+        return [{"error": str(exc)}]
+    except Exception as exc:  # noqa: BLE001
+        logger.error("search_meetings error: %s", exc)
+        return [{"error": str(exc)}]
+
+    documents = (results.get("documents") or [[]])[0]
+    metadatas = (results.get("metadatas") or [[]])[0]
+    distances = (results.get("distances") or [[]])[0]
+
+    items: List[Dict[str, Any]] = []
+    for i, doc in enumerate(documents):
+        meta = metadatas[i] if i < len(metadatas) else {}
+        distance = distances[i] if i < len(distances) else None
+        # Chroma's default space is squared-L2 (unbounded above). Map to a
+        # bounded, monotonically decreasing relevance score in (0, 1].
+        score = (
+            round(1.0 / (1.0 + float(distance)), 6) if distance is not None else None
+        )
+        snippet = (doc or "")[:280]
+        items.append(
+            {
+                "meeting_id": meta.get("meeting_id"),
+                "score": score,
+                "snippet": snippet,
+                "timestamp_end": meta.get("timestamp_end", 0.0),
+                "timestamp_start": meta.get("timestamp_start", 0.0),
+                "title": meta.get("title"),
+            }
+        )
+    return items

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "praisonai-tools"
-version = "0.2.37"
+version = "0.2.41"
 description = "Extended tools for PraisonAI Agents"
 authors = [
     {name = "Mervin Praison"}
@@ -80,6 +80,10 @@ video-motion = [
 composio = [
     "composio>=0.5.0",
 ]
+meeting = [
+    "openai>=1.12.0",
+    "chromadb>=0.4.0",
+]
 
 [project.urls]
 Homepage = "https://docs.praison.ai"

tests/unit/tools/test_meeting_tools.py

@@ -0,0 +1,387 @@
+"""Unit tests for meeting_tools.
+
+All external clients (OpenAI, Chroma) are mocked so no network or heavy
+dependency is required. The SQLite persistence is redirected to a temp dir.
+"""
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from praisonai_tools.tools import meeting_tools as mt
+
+
+@pytest.fixture(autouse=True)
+def _isolated_app_dir(tmp_path, monkeypatch):
+    """Point the meetings DB and vector store at a temp dir per test."""
+    monkeypatch.setenv("PRAISONAI_MEETINGS_DIR", str(tmp_path))
+    monkeypatch.setenv("PRAISONAI_MEETINGS_DB", str(tmp_path / "meetings.db"))
+    monkeypatch.setenv("OPENAI_API_KEY", "test-key")
+    yield
+
+
+# ── transcribe_file ─────────────────────────────────────────────────
+
+
+class TestTranscribeFile:
+    def test_missing_file(self):
+        result = mt.transcribe_file.__wrapped__("/no/such/file.mp3")
+        assert "error" in result
+        assert "not found" in result["error"].lower()
+
+    def _fake_response(self):
+        return SimpleNamespace(
+            text="Good morning everyone",
+            duration=12.5,
+            language="en",
+            segments=[
+                SimpleNamespace(start=0.0, end=4.2, text="Good morning everyone"),
+            ],
+        )
+
+    def test_success(self, tmp_path):
+        audio = tmp_path / "a.mp3"
+        audio.write_bytes(b"fake")
+        client = MagicMock()
+        client.audio.transcriptions.create.return_value = self._fake_response()
+        with patch.object(mt, "_get_openai_client", return_value=client):
+            result = mt.transcribe_file.__wrapped__(str(audio))
+        assert result["text"] == "Good morning everyone"
+        assert result["language"] == "en"
+        assert result["duration_seconds"] == 12.5
+        assert result["segments"][0]["start"] == 0.0
+        assert result["segments"][0]["end"] == 4.2
+
+    def test_403_error_mapping(self, tmp_path):
+        audio = tmp_path / "a.mp3"
+        audio.write_bytes(b"fake")
+        exc = Exception("403 forbidden")
+        setattr(exc, "status_code", 403)
+        client = MagicMock()
+        client.audio.transcriptions.create.side_effect = exc
+        with patch.object(mt, "_get_openai_client", return_value=client):
+            result = mt.transcribe_file.__wrapped__(str(audio))
+        assert "403" in result["error"] or "permission" in result["error"].lower()
+
+    def test_timeout_retries_once_then_fails(self, tmp_path):
+        audio = tmp_path / "a.mp3"
+        audio.write_bytes(b"fake")
+
+        class TimeoutError_(Exception):
+            pass
+
+        client = MagicMock()
+        client.audio.transcriptions.create.side_effect = TimeoutError_("timeout")
+        with patch.object(mt, "_get_openai_client", return_value=client):
+            result = mt.transcribe_file.__wrapped__(str(audio))
+        assert "timed out" in result["error"].lower()
+        assert client.audio.transcriptions.create.call_count == 2
+
+    def test_empty_audio(self, tmp_path):
+        audio = tmp_path / "a.mp3"
+        audio.write_bytes(b"fake")
+        client = MagicMock()
+        client.audio.transcriptions.create.return_value = SimpleNamespace(
+            text="", duration=0.0, language="en", segments=[]
+        )
+        with patch.object(mt, "_get_openai_client", return_value=client):
+            result = mt.transcribe_file.__wrapped__(str(audio))
+        assert result["error"] == "No speech detected in recording"
+
+
+# ── Meetings CRUD ───────────────────────────────────────────────────
+
+
+class TestMeetingsCRUD:
+    def test_save_and_get(self):
+        saved = mt.save_meeting.__wrapped__(
+            title="Weekly standup", source="file.mp3", metadata={"team": "eng"}
+        )
+        mid = saved["meeting_id"]
+        assert mid
+        record = mt.get_meeting.__wrapped__(mid)
+        assert record["title"] == "Weekly standup"
+        assert record["source"] == "file.mp3"
+        assert record["metadata"] == {"team": "eng"}
+
+    def test_save_requires_title(self):
+        assert "error" in mt.save_meeting.__wrapped__(title="")
+
+    def test_get_not_found(self):
+        assert "error" in mt.get_meeting.__wrapped__("nope")
+
+    def test_list(self):
+        mt.save_meeting.__wrapped__(title="A")
+        mt.save_meeting.__wrapped__(title="B")
+        meetings = mt.list_meetings.__wrapped__(limit=10, offset=0)
+        assert len(meetings) == 2
+        titles = {m["title"] for m in meetings}
+        assert titles == {"A", "B"}
+
+    def test_list_pagination(self):
+        for i in range(3):
+            mt.save_meeting.__wrapped__(title=f"M{i}")
+        page = mt.list_meetings.__wrapped__(limit=1, offset=1)
+        assert len(page) == 1
+
+
+# ── summarize_transcript ────────────────────────────────────────────
+
+
+class TestSummarize:
+    def test_requires_transcript(self):
+        assert "error" in mt.summarize_transcript.__wrapped__("")
+
+    def test_single_chunk(self):
+        payload = {
+            "summary": "It was a good meeting.",
+            "decisions": ["Ship on Friday"],
+            "topics": ["release"],
+            "key_quotes": [{"speaker": "Alex", "quote": "Ship it"}],
+        }
+        with patch.object(mt, "_chat_json", return_value=payload):
+            result = mt.summarize_transcript.__wrapped__("short transcript")
+        assert result["summary"] == "It was a good meeting."
+        assert result["decisions"] == ["Ship on Friday"]
+        assert result["topics"] == ["release"]
+        assert result["key_quotes"][0]["quote"] == "Ship it"
+
+    def test_map_reduce_merges_chunks(self):
+        long_text = "word " * 8000  # forces multiple summary chunks
+        calls = []
+
+        def fake_chat(system, user):
+            idx = len(calls)
+            calls.append(user)
+            return {
+                "summary": f"para{idx}",
+                "decisions": [f"d{idx}"],
+                "topics": ["t"],
+                "key_quotes": [{"speaker": None, "quote": f"q{idx}"}],
+            }
+
+        with patch.object(mt, "_chat_json", side_effect=fake_chat):
+            result = mt.summarize_transcript.__wrapped__(long_text)
+
+        assert len(calls) > 1
+        assert "para0" in result["summary"]
+        assert "t" in result["topics"]  # deduped
+        assert result["topics"].count("t") == 1
+        assert len(result["decisions"]) == len(calls)
+
+
+# ── extract_action_items ────────────────────────────────────────────
+
+
+class TestActionItems:
+    def test_requires_transcript(self):
+        assert "error" in mt.extract_action_items.__wrapped__("")
+
+    def test_returns_valid_schema(self):
+        payload = {
+            "action_items": [
+                {
+                    "description": "Fix API bug",
+                    "owner": "Alex",
+                    "due_date": "2026-09-05",
+                    "priority": "high",
+                }
+            ]
+        }
+        with patch.object(mt, "_chat_json", return_value=payload):
+            result = mt.extract_action_items.__wrapped__("transcript")
+        item = result["action_items"][0]
+        assert set(item.keys()) == {"description", "owner", "due_date", "priority"}
+        assert item["owner"] == "Alex"
+
+    def test_handles_empty(self):
+        with patch.object(mt, "_chat_json", return_value={}):
+            result = mt.extract_action_items.__wrapped__("transcript")
+        assert result["action_items"] == []
+
+
+# ── Fake in-memory vector collection ────────────────────────────────
+
+
+class _FakeCollection:
+    """Minimal Chroma-like collection using cosine-ish distance on tags."""
+
+    def __init__(self):
+        self.store = {}  # id -> (document, metadata, embedding)
+
+    def delete(self, where=None):
+        if not where:
+            self.store.clear()
+            return
+        mid = where.get("meeting_id")
+        for k in [k for k, v in self.store.items() if v[1].get("meeting_id") == mid]:
+            del self.store[k]
+
+    def add(self, ids, documents, embeddings, metadatas):
+        for i, _id in enumerate(ids):
+            self.store[_id] = (documents[i], metadatas[i], embeddings[i])
+
+    def query(self, query_embeddings, n_results):
+        q = query_embeddings[0]
+
+        def dist(emb):
+            return sum((a - b) ** 2 for a, b in zip(q, emb))
+
+        ranked = sorted(self.store.items(), key=lambda kv: dist(kv[1][2]))
+        ranked = ranked[:n_results]
+        return {
+            "documents": [[v[0] for _, v in ranked]],
+            "metadatas": [[v[1] for _, v in ranked]],
+            "distances": [[dist(v[2]) for _, v in ranked]],
+        }
+
+
+class TestIndexAndSearch:
+    def _embed_map(self):
+        # deterministic embeddings keyed by content keyword
+        def fake_embed(texts):
+            out = []
+            for t in texts:
+                if "apple" in t.lower():
+                    out.append([1.0, 0.0])
+                elif "banana" in t.lower():
+                    out.append([0.0, 1.0])
+                else:
+                    out.append([0.5, 0.5])
+            return out
+
+        return fake_embed
+
+    def test_index_and_rank(self):
+        coll = _FakeCollection()
+        m1 = mt.save_meeting.__wrapped__(title="Apple meeting")["meeting_id"]
+        m2 = mt.save_meeting.__wrapped__(title="Banana meeting")["meeting_id"]
+        with patch.object(mt, "_get_vector_collection", return_value=coll), patch.object(
+            mt, "_embed", side_effect=self._embed_map()
+        ):
+            mt.index_meeting.__wrapped__(m1, transcript="all about apple pie")
+            mt.index_meeting.__wrapped__(m2, transcript="all about banana bread")
+            results = mt.search_meetings.__wrapped__("apple", limit=2)
+
+        assert results[0]["meeting_id"] == m1
+        assert results[0]["score"] is not None
+        assert results[0]["title"] == "Apple meeting"
+
+    def test_reindex_does_not_duplicate(self):
+        coll = _FakeCollection()
+        mid = mt.save_meeting.__wrapped__(title="M")["meeting_id"]
+        with patch.object(mt, "_get_vector_collection", return_value=coll), patch.object(
+            mt, "_embed", side_effect=self._embed_map()
+        ):
+            r1 = mt.index_meeting.__wrapped__(mid, transcript="apple " * 600)
+            count_after_first = len(coll.store)
+            r2 = mt.index_meeting.__wrapped__(mid, transcript="apple " * 600)
+            count_after_second = len(coll.store)
+
+        assert r1["chunks_indexed"] == r2["chunks_indexed"]
+        assert count_after_first == count_after_second
+
+    def test_index_missing_transcript(self):
+        mid = mt.save_meeting.__wrapped__(title="M")["meeting_id"]
+        result = mt.index_meeting.__wrapped__(mid)
+        assert "error" in result
+
+    def test_search_requires_query(self):
+        assert "error" in mt.search_meetings.__wrapped__("")[0]
+
+    def test_segments_map_to_chunk_timestamps(self):
+        coll = _FakeCollection()
+        mid = mt.save_meeting.__wrapped__(title="Timed")["meeting_id"]
+        segments = [
+            {"start": 0.0, "end": 5.0, "text": "apple " * 200},
+            {"start": 5.0, "end": 12.0, "text": "banana " * 200},
+        ]
+        transcript = " ".join(s["text"].strip() for s in segments)
+        with patch.object(mt, "_get_vector_collection", return_value=coll), patch.object(
+            mt, "_embed", side_effect=self._embed_map()
+        ):
+            mt.index_meeting.__wrapped__(mid, transcript=transcript, segments=segments)
+            results = mt.search_meetings.__wrapped__("banana", limit=1)
+
+        assert results[0]["meeting_id"] == mid
+        # The best match ("banana") must carry a non-zero timestamp derived
+        # from the second segment, not the recording start.
+        assert results[0]["timestamp_start"] > 0.0
+        assert results[0]["timestamp_end"] >= results[0]["timestamp_start"]
+
+    def test_failed_reindex_preserves_existing_chunks(self):
+        coll = _FakeCollection()
+        mid = mt.save_meeting.__wrapped__(title="M")["meeting_id"]
+        with patch.object(mt, "_get_vector_collection", return_value=coll), patch.object(
+            mt, "_embed", side_effect=self._embed_map()
+        ):
+            mt.index_meeting.__wrapped__(mid, transcript="apple " * 600)
+        assert len(coll.store) > 0
+        original = dict(coll.store)
+
+        def boom(_texts):
+            raise RuntimeError("embedding backend down")
+
+        with patch.object(mt, "_get_vector_collection", return_value=coll), patch.object(
+            mt, "_embed", side_effect=boom
+        ):
+            result = mt.index_meeting.__wrapped__(mid, transcript="apple " * 600)
+
+        assert "error" in result
+        # Previous chunks must survive a failed re-index.
+        assert coll.store == original
+
+    def test_search_score_bounded_for_large_distance(self):
+        coll = _FakeCollection()
+        mid = mt.save_meeting.__wrapped__(title="Far")["meeting_id"]
+
+        def far_embed(texts):
+            # Query and doc are far apart -> squared-L2 distance > 1.
+            return [[10.0, 10.0] for _ in texts]
+
+        with patch.object(mt, "_get_vector_collection", return_value=coll), patch.object(
+            mt, "_embed", side_effect=lambda t: [[0.0, 0.0] for _ in t]
+        ):
+            mt.index_meeting.__wrapped__(mid, transcript="apple pie")
+        with patch.object(mt, "_get_vector_collection", return_value=coll), patch.object(
+            mt, "_embed", side_effect=far_embed
+        ):
+            results = mt.search_meetings.__wrapped__("pie", limit=1)
+
+        assert 0.0 < results[0]["score"] <= 1.0
+
+
+# ── Lazy import hygiene ─────────────────────────────────────────────
+
+
+def test_no_heavy_deps_at_import_time():
+    import sys
+
+    # meeting_tools is already imported at top; ensure it did not pull heavy deps.
+    mod = sys.modules["praisonai_tools.tools.meeting_tools"]
+    src = open(mod.__file__).read()
+    # No top-level (unindented) imports of openai/chromadb.
+    for line in src.splitlines():
+        if line.startswith("import openai") or line.startswith("from openai"):
+            raise AssertionError("openai imported at module top level")
+        if line.startswith("import chromadb") or line.startswith("from chromadb"):
+            raise AssertionError("chromadb imported at module top level")
+
+
+def test_all_tools_are_registerable():
+    import praisonai_tools
+
+    for name in (
+        "transcribe_file",
+        "save_meeting",
+        "get_meeting",
+        "list_meetings",
+        "summarize_transcript",
+        "extract_action_items",
+        "index_meeting",
+        "search_meetings",
+    ):
+        exported = getattr(praisonai_tools, name, None)
+        assert exported is not None, f"{name} is not exported"
+        assert exported is getattr(mt, name), f"{name} export is shadowed"