orchestrata_ultimate.integrations.fireflies_client

Reference › orchestrata_ultimate › integrations › fireflies_client

fireflies_client is the read boundary to the external Fireflies service (specification 018). FirefliesClient is an async GraphQL client built on httpx. The gateway's Fireflies handler imports it, and fireflies_sync builds on it. The curated narrative is integrations.

The constructor takes an API key, an endpoint, an optional injected HTTP client for testing and an optional environment mapping; without an explicit key it reads the FIREFLIES_API_KEY environment variable, and without any key the methods return empty, well-shaped results instead of raising. _request posts one GraphQL query and returns the JSON. The public methods list recent records, fetch one in full, search, and fetch an AI-generated summary; test_connection validates the configured key with a cheap probe, and close releases the HTTP client when the instance created it. On an httpx error or a GraphQL error payload each method returns the same shape with an added error key, so callers do not need exception handling. Every public method carries a @traces marker for specification 018.

At a glance

Class diagram

Dependency diagram

API

Classes

FirefliesClient

Line 88

Async GraphQL client for Fireflies meeting intelligence API.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/integrations/fireflies_client.py

"""Async GraphQL client for the Fireflies API.

Provides methods for listing, searching, and summarizing meeting transcripts
via Fireflies' GraphQL endpoint, plus a cheap connection-validation probe.
"""

import logging
import os
from collections.abc import Mapping
from typing import Any

import httpx

from orchestrata_ultimate.core.base.traces import traces

logger = logging.getLogger(__name__)

API_ENDPOINT = "https://api.fireflies.ai/graphql"

_LIST_TRANSCRIPTS_QUERY = """
query Transcripts($limit: Int, $offset: Int) {
  transcripts(limit: $limit, offset: $offset) {
    id
    title
    date
    duration
    organizer_email
    participants
  }
}
"""

_GET_TRANSCRIPT_QUERY = """
query Transcript($id: String!) {
  transcript(id: $id) {
    id
    title
    date
    duration
    organizer_email
    participants
    sentences {
      speaker_name
      text
      start_time
      end_time
    }
  }
}
"""

_SEARCH_TRANSCRIPTS_QUERY = """
query Search($query: String!, $limit: Int, $offset: Int) {
  search(query: $query, limit: $limit, offset: $offset) {
    transcript_id
    title
    highlights
    score
  }
}
"""

_SUMMARIZE_TRANSCRIPT_QUERY = """
query Summary($id: String!) {
  transcript(id: $id) {
    id
    summary {
      overview
      action_items
      topics
      questions
      follow_ups
    }
  }
}
"""

_TEST_CONNECTION_QUERY = """
query {
  user {
    user_id
    email
  }
}
"""


class FirefliesClient:
    """Async GraphQL client for Fireflies meeting intelligence API.

    Supports transcript list/search/summary operations and a lightweight
    ``test_connection()`` probe that validates the configured API key.
    """

    @traces(spec="018", story="US-006", reqs=["REQ-018-073"])
    def __init__(
        self,
        api_key: str | None = None,
        endpoint: str = API_ENDPOINT,
        http_client: httpx.AsyncClient | None = None,
        env: Mapping[str, str] | None = None,
    ) -> None:
        """Initialize the Fireflies client.

        Args:
            api_key: Fireflies API key. Falls back to ``FIREFLIES_API_KEY``
                in ``env`` (or ``os.environ`` if ``env`` is not provided)
                if not provided.
            endpoint: GraphQL endpoint URL.
            http_client: Optional ``httpx.AsyncClient`` instance for testing.
            env: Optional mapping of environment variables. Defaults to
                ``os.environ``.
        """
        self._api_key = api_key or (env or os.environ).get("FIREFLIES_API_KEY") or ""
        self._endpoint = endpoint
        self._client = http_client
        self._owns_client = http_client is None

    @traces(spec="018", story="US-006", reqs=["REQ-018-073"])
    async def _request(self, query: str, variables: dict[str, Any]) -> dict[str, Any]:
        """Execute a GraphQL request and return the JSON response.

        Args:
            query: GraphQL query string.
            variables: Variables for the query.

        Returns:
            Parsed JSON response dictionary.
        """
        if not self._client:
            self._client = httpx.AsyncClient(
                headers={
                    "Authorization": f"Bearer {self._api_key}",
                    "Content-Type": "application/json",
                },
                timeout=30.0,
            )

        payload = {"query": query, "variables": variables}
        response = await self._client.post(self._endpoint, json=payload)
        response.raise_for_status()
        data: dict[str, Any] = response.json()
        return data

    @traces(spec="018", story="US-006", reqs=["REQ-018-066"])
    async def list_transcripts(
        self, limit: int = 20, offset: int = 0
    ) -> dict[str, Any]:
        """List recent meeting transcripts.

        Args:
            limit: Maximum number of transcripts to return.
            offset: Offset for pagination.

        Returns:
            Dictionary with ``transcripts``, ``count``, ``limit``, and
            ``offset`` keys. ``count`` is the number of transcripts in this
            page (Fireflies' GraphQL API exposes no corpus-wide total-count
            field, so this is honestly the page size, not a total). Contains
            an ``error`` key on failure.
        """
        if not self._api_key:
            return {
                "transcripts": [],
                "count": 0,
                "limit": limit,
                "offset": offset,
            }

        try:
            data = await self._request(
                _LIST_TRANSCRIPTS_QUERY,
                {"limit": limit, "offset": offset},
            )
            if "errors" in data:
                error_msg = data["errors"][0].get("message", "GraphQL error")
                logger.warning("Fireflies list_transcripts error: %s", error_msg)
                return {
                    "error": error_msg,
                    "transcripts": [],
                    "count": 0,
                    "limit": limit,
                    "offset": offset,
                }

            transcripts = data.get("data", {}).get("transcripts", [])
            return {
                "transcripts": transcripts,
                "count": len(transcripts),
                "limit": limit,
                "offset": offset,
            }
        except httpx.HTTPError as exc:
            logger.warning("Fireflies list_transcripts HTTP error: %s", exc)
            return {
                "error": str(exc),
                "transcripts": [],
                "count": 0,
                "limit": limit,
                "offset": offset,
            }

    @traces(spec="018", story="US-006", reqs=["REQ-018-067"])
    async def get_transcript(self, transcript_id: str | None) -> dict[str, Any]:
        """Fetch a single transcript with full details.

        Args:
            transcript_id: The Fireflies transcript ID.

        Returns:
            Dictionary with transcript details. Contains an ``error`` key
            on failure.
        """
        if not self._api_key:
            return {
                "transcript_id": transcript_id,
                "title": "",
                "date": None,
                "duration": 0,
                "participants": [],
                "text": "",
                "speakers": [],
                "segments": [],
            }

        try:
            data = await self._request(
                _GET_TRANSCRIPT_QUERY,
                {"id": transcript_id},
            )
            if "errors" in data:
                error_msg = data["errors"][0].get("message", "GraphQL error")
                logger.warning("Fireflies get_transcript error: %s", error_msg)
                return {
                    "error": error_msg,
                    "transcript_id": transcript_id,
                    "title": "",
                    "date": None,
                    "duration": 0,
                    "participants": [],
                    "text": "",
                    "segments": [],
                }

            transcript = data.get("data", {}).get("transcript", {})
            sentences = transcript.get("sentences", [])
            segments = [
                {
                    "speaker": s.get("speaker_name", ""),
                    "text": s.get("text", ""),
                    "start_time": s.get("start_time", 0),
                    "end_time": s.get("end_time", 0),
                }
                for s in sentences
            ]
            full_text = " ".join(s.get("text", "") for s in sentences)
            speakers = list(
                {s.get("speaker_name", "") for s in sentences if s.get("speaker_name")}
            )

            return {
                "transcript_id": transcript.get("id", transcript_id),
                "title": transcript.get("title", ""),
                "date": transcript.get("date"),
                "duration": transcript.get("duration", 0),
                "participants": transcript.get("participants", []),
                "text": full_text,
                "speakers": speakers,
                "segments": segments,
            }
        except httpx.HTTPError as exc:
            logger.warning("Fireflies get_transcript HTTP error: %s", exc)
            return {
                "error": str(exc),
                "transcript_id": transcript_id,
                "title": "",
                "date": None,
                "duration": 0,
                "participants": [],
                "text": "",
                "speakers": [],
                "segments": [],
            }

    @traces(spec="018", story="US-006", reqs=["REQ-018-068"])
    async def search_transcripts(
        self, query: str, limit: int = 20, offset: int = 0
    ) -> dict[str, Any]:
        """Search across meeting transcripts.

        Args:
            query: Search query string.
            limit: Maximum number of results.
            offset: Offset for pagination.

        Returns:
            Dictionary with ``results``, ``count``, ``query``, ``limit``,
            and ``offset`` keys. ``count`` is the number of results in this
            page (Fireflies' GraphQL API exposes no corpus-wide total-count
            field, so this is honestly the page size, not a total). Contains
            an ``error`` key on failure.
        """
        if not self._api_key:
            return {
                "results": [],
                "count": 0,
                "query": query,
                "limit": limit,
                "offset": offset,
            }

        try:
            data = await self._request(
                _SEARCH_TRANSCRIPTS_QUERY,
                {"query": query, "limit": limit, "offset": offset},
            )
            if "errors" in data:
                error_msg = data["errors"][0].get("message", "GraphQL error")
                logger.warning("Fireflies search_transcripts error: %s", error_msg)
                return {
                    "error": error_msg,
                    "results": [],
                    "count": 0,
                    "query": query,
                    "limit": limit,
                    "offset": offset,
                }

            results = data.get("data", {}).get("search", [])
            return {
                "results": results,
                "count": len(results),
                "query": query,
                "limit": limit,
                "offset": offset,
            }
        except httpx.HTTPError as exc:
            logger.warning("Fireflies search_transcripts HTTP error: %s", exc)
            return {
                "error": str(exc),
                "results": [],
                "count": 0,
                "query": query,
                "limit": limit,
                "offset": offset,
            }

    @traces(spec="018", story="US-006", reqs=["REQ-018-069"])
    async def summarize_transcript(self, transcript_id: str | None) -> dict[str, Any]:
        """Fetch an AI-generated summary for a transcript.

        Args:
            transcript_id: The Fireflies transcript ID.

        Returns:
            Dictionary with ``topics``, ``action_items``, ``questions``,
            ``follow_ups``, and ``generated_at`` keys. ``questions`` holds
            the questions raised during the meeting, as returned by
            Fireflies' GraphQL ``summary.questions`` field -- the API has no
            "decisions" field, so this method does not fabricate one.
            Contains an ``error`` key on failure.
        """
        if not self._api_key:
            return {
                "transcript_id": transcript_id,
                "topics": [],
                "action_items": [],
                "questions": [],
                "follow_ups": [],
                "generated_at": None,
            }

        try:
            data = await self._request(
                _SUMMARIZE_TRANSCRIPT_QUERY,
                {"id": transcript_id},
            )
            if "errors" in data:
                error_msg = data["errors"][0].get("message", "GraphQL error")
                logger.warning("Fireflies summarize_transcript error: %s", error_msg)
                return {
                    "error": error_msg,
                    "transcript_id": transcript_id,
                    "topics": [],
                    "action_items": [],
                    "questions": [],
                    "follow_ups": [],
                    "generated_at": None,
                }

            transcript = data.get("data", {}).get("transcript", {})
            summary = transcript.get("summary", {})

            return {
                "transcript_id": transcript.get("id", transcript_id),
                "topics": summary.get("topics", []),
                "action_items": summary.get("action_items", []),
                "questions": summary.get("questions", []),
                "follow_ups": summary.get("follow_ups", []),
                "generated_at": None,
            }
        except httpx.HTTPError as exc:
            logger.warning("Fireflies summarize_transcript HTTP error: %s", exc)
            return {
                "error": str(exc),
                "transcript_id": transcript_id,
                "topics": [],
                "action_items": [],
                "questions": [],
                "follow_ups": [],
                "generated_at": None,
            }

    @traces(spec="018", story="US-006", reqs=["REQ-018-058", "REQ-018-080"])
    async def test_connection(self) -> dict[str, Any]:
        """Validate the Fireflies API key with a cheap GraphQL probe.

        Returns:
            ``{"ok": True, "user": {...}}`` on success, or
            ``{"ok": False, "error": "..."}`` on failure. Never raises.
        """
        if not self._api_key:
            return {
                "ok": False,
                "error": "FIREFLIES_API_KEY not configured",
            }

        try:
            data = await self._request(_TEST_CONNECTION_QUERY, {})
            if "errors" in data:
                error_msg = data["errors"][0].get("message", "GraphQL error")
                logger.warning("Fireflies test_connection error: %s", error_msg)
                return {
                    "ok": False,
                    "error": error_msg,
                }

            user = data.get("data", {}).get("user", {})
            return {
                "ok": True,
                "user": user,
            }
        except httpx.HTTPError as exc:
            logger.warning("Fireflies test_connection HTTP error: %s", exc)
            return {
                "ok": False,
                "error": str(exc),
            }

    async def close(self) -> None:
        """Close the underlying HTTP client."""
        if self._client and self._owns_client:
            await self._client.aclose()
            self._client = None