orchestrata_ultimate.gateway.handlers.fireflies

Reference › orchestrata_ultimate › gateway › handlers › fireflies

FirefliesHandler is the gateway's bridge to the Fireflies integration in the integrations package. It serves the fireflies.* family of RPCs: listing, single-item lookup, search, summaries, a health check and a sync call. rpc creates it. The handler builds a FirefliesClient on demand with _get_client and passes most calls straight through to it; handle_health returns the client's connection check. handle_sync validates a limit between 1 and 100, returns an error if no graph repository was attached with set_graph_repository, and otherwise runs FirefliesKnowledgeSync.sync_recent from integrations.fireflies_sync, which writes recent items into the knowledge graph and returns a report with the synced count, node ids and errors. See gateway.

At a glance

Class diagram

Dependency diagram

API

Classes

FirefliesHandler

Line 17

Fireflies meeting intelligence and transcript methods.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/gateway/handlers/fireflies.py

"""Fireflies RPC handlers -- transcript retrieval, search, summaries, and health.

Implements SPEC-018 FR-066 through FR-069 MCP bridge methods plus
``fireflies.health`` connection validation and ``fireflies.sync`` KG sync.
"""

import logging
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.integrations.fireflies_client import FirefliesClient
from orchestrata_ultimate.integrations.fireflies_sync import FirefliesKnowledgeSync

logger = logging.getLogger(__name__)


class FirefliesHandler:
    """Fireflies meeting intelligence and transcript methods.

    Provides RPC methods for fetching meeting transcripts, searching
    across transcripts, retrieving AI-generated summaries, validating the
    Fireflies connection, and syncing transcripts into the knowledge graph.
    """

    def __init__(self) -> None:
        """Initialize Fireflies handler with MCP bridge capabilities."""
        self._client: FirefliesClient | None = None
        self._graph_repository: Any = None

    @traces(spec="018", story="US-006", reqs=["REQ-018-073"])
    def _get_client(self, api_key: str | None = None) -> FirefliesClient:
        """Return a Fireflies client, creating one if needed."""
        if self._client is None:
            self._client = FirefliesClient(api_key=api_key)
        return self._client

    def set_graph_repository(self, graph_repository: Any) -> None:
        """Inject the operational-graph repository for sync operations.

        Args:
            graph_repository: An AsyncGraphRepository instance.
        """
        self._graph_repository = graph_repository

    @traces(spec="018", story="US-006", reqs=["REQ-018-066"])
    async def handle_transcripts(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle fireflies.transcripts -- fetch list of recent meeting transcripts.

        Args:
            connection_id: Caller connection ID.
            params: Optional limit and offset for pagination.

        Returns:
            List of transcripts with metadata.
        """
        limit = params.get("limit", 20)
        offset = params.get("offset", 0)
        api_key = params.get("api_key")

        client = self._get_client(api_key=api_key)
        return await client.list_transcripts(limit=limit, offset=offset)

    @traces(spec="018", story="US-006", reqs=["REQ-018-067"])
    async def handle_transcript(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle fireflies.transcript -- fetch a single transcript with full text.

        Args:
            connection_id: Caller connection ID.
            params: Transcript ID to retrieve.

        Returns:
            Full transcript with speaker segments and timestamps.
        """
        transcript_id = params.get("transcript_id")
        api_key = params.get("api_key")

        client = self._get_client(api_key=api_key)
        return await client.get_transcript(transcript_id=transcript_id)

    @traces(spec="018", story="US-006", reqs=["REQ-018-068"])
    async def handle_search(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle fireflies.search -- search across all meeting transcripts.

        Args:
            connection_id: Caller connection ID.
            params: Search query, limit, and offset.

        Returns:
            Search results with highlighted excerpts.
        """
        query = params.get("query", "")
        limit = params.get("limit", 20)
        offset = params.get("offset", 0)
        api_key = params.get("api_key")

        client = self._get_client(api_key=api_key)
        return await client.search_transcripts(query=query, limit=limit, offset=offset)

    @traces(spec="018", story="US-006", reqs=["REQ-018-069"])
    async def handle_summary(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle fireflies.summary -- fetch AI-generated meeting summary.

        Args:
            connection_id: Caller connection ID.
            params: Transcript ID to summarize.

        Returns:
            Meeting summary with topics, action items, questions, and follow-ups.
        """
        transcript_id = params.get("transcript_id")
        api_key = params.get("api_key")

        client = self._get_client(api_key=api_key)
        return await client.summarize_transcript(transcript_id=transcript_id)

    @traces(spec="018", story="US-006", reqs=["REQ-018-080"])
    async def handle_health(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle fireflies.health -- validate Fireflies API connectivity.

        Args:
            connection_id: Caller connection ID.
            params: Optional ``api_key`` to test a specific key.

        Returns:
            Connection check result with ``ok`` and ``user`` keys, or an
            ``ok``/``error`` payload on failure.
        """
        api_key = params.get("api_key")
        client = self._get_client(api_key=api_key)
        return await client.test_connection()

    @traces(spec="018", story="US-006", reqs=["REQ-018-012"])
    async def handle_sync(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle fireflies.sync -- pull recent transcripts into the knowledge graph.

        Args:
            connection_id: Caller connection ID.
            params: Optional ``limit`` (int, 1..100, default 20) and ``api_key``.

        Returns:
            Sync report with ``synced``, ``node_ids``, and ``errors`` keys,
            or an ``error`` payload if validation fails or the graph is unavailable.
        """
        limit = params.get("limit", 20)
        if not isinstance(limit, int) or not (1 <= limit <= 100):
            return {
                "error": "limit must be an integer between 1 and 100",
            }

        if self._graph_repository is None:
            return {
                "error": "Graph unavailable",
            }

        api_key = params.get("api_key")
        client = self._get_client(api_key=api_key)
        sync = FirefliesKnowledgeSync(
            client=client,
            repo=self._graph_repository,
        )
        return await sync.sync_recent(limit=limit)