orchestrata_ultimate.session.fireflies_tools
Reference › orchestrata_ultimate › session › fireflies_tools
FirefliesMCPTools is an MCP tool provider for the Fireflies service integration. Its constructor takes an optional API key and a cache lifetime in seconds (default 600). tool_definitions returns the MCP schemas, and handle_tool_call(tool_name, arguments) routes a call to one of five handlers: listing records, fetching one record, searching, fetching a summary, and testing the connection. It follows the same provider shape as SessionMCPTools.
At a glance
| Source | src/orchestrata_ultimate/session/fireflies_tools.py |
| Lines of code | 237 |
| Classes | 1 |
| Functions | 0 |
| Imports (internal) | orchestrata_ultimate.core.base.traces, orchestrata_ultimate.integrations.fireflies_client |
| Imported by | none |
| Tests that import it | tests/generated_recovery/test_session_fireflies_tools.py, tests/session/test_fireflies_tools.py, tests/session/test_session_fireflies_tools.py, tests/session/test_session_signatures.py |
Class diagram
Dependency diagram
API
Classes
FirefliesMCPTools
Decorators: traces(spec='018', story='US-006', reqs=['REQ-018-066', 'REQ-018-067', 'REQ-018-068', 'REQ-018-069', 'REQ-018-070', 'RE… · Line 38
MCP tool provider for Fireflies meeting intelligence operations.
| Method | Signature | Summary |
|---|---|---|
__init__ |
__init__(api_key: str | None = None, cache_ttl: int = 600) -> None |
Initialize Fireflies MCP tools. |
tool_definitions property |
tool_definitions() -> list[dict[str, Any]] |
Return MCP tool schema definitions. |
handle_tool_call |
handle_tool_call(tool_name: str, arguments: dict[str, Any]) -> dict[str, Any] |
Route an MCP tool call to the appropriate handler. |
_get_transcripts |
_get_transcripts(args: dict[str, Any]) -> dict[str, Any] |
Get list of recent transcripts. |
_get_transcript |
_get_transcript(args: dict[str, Any]) -> dict[str, Any] |
Get single transcript with full text. |
_search |
_search(args: dict[str, Any]) -> dict[str, Any] |
Search across transcripts. |
_get_summary |
_get_summary(args: dict[str, Any]) -> dict[str, Any] |
Get AI-generated meeting summary. |
_test_connection |
_test_connection(args: dict[str, Any]) -> dict[str, Any] |
Validate Fireflies API connectivity. |
Source
"""Fireflies MCP tool definitions for meeting intelligence.
Provides tool schemas and handler functions that expose Fireflies
meeting intelligence operations in MCP-compatible format.
These can be registered with any MCP server implementation or used
by the Fireflies MCP bridge plugin.
No dependency on FastMCP or any external MCP library -- provides a clean
interface that any MCP server can wrap.
Implements SPEC-018 US-006 MCP Bridge requirements.
"""
import asyncio
import logging
from typing import Any
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.integrations.fireflies_client import FirefliesClient
logger = logging.getLogger(__name__)
@traces(
spec="018",
story="US-006",
reqs=[
"REQ-018-066",
"REQ-018-067",
"REQ-018-068",
"REQ-018-069",
"REQ-018-070",
"REQ-018-074",
"REQ-018-075",
],
)
class FirefliesMCPTools:
"""MCP tool provider for Fireflies meeting intelligence operations.
Exposes transcript retrieval, search, summary, and connection-validation
operations as MCP-compatible tools with standardized schemas.
"""
@traces(spec="018", story="US-006", reqs=["REQ-018-074"])
def __init__(self, api_key: str | None = None, cache_ttl: int = 600) -> None:
"""Initialize Fireflies MCP tools.
Args:
api_key: Fireflies API key for authentication.
cache_ttl: Cache TTL in seconds (default 10 minutes for REQ-018-074).
"""
self._api_key = api_key
self._cache_ttl = cache_ttl
self._cache: dict[str, tuple[Any, float]] = {}
self._client = FirefliesClient(api_key=api_key)
@property
@traces(
spec="018",
story="US-006",
reqs=["REQ-018-066", "REQ-018-067", "REQ-018-068", "REQ-018-069"],
)
def tool_definitions(self) -> list[dict[str, Any]]:
"""Return MCP tool schema definitions.
Each tool follows the MCP tool schema format with name,
description, and JSON Schema inputSchema.
"""
return [
{
"name": "fireflies_get_transcripts",
"description": "Get a list of recent meeting transcripts from Fireflies",
"inputSchema": {
"type": "object",
"properties": {
"limit": {
"type": "integer",
"description": "Maximum number of transcripts to return",
"default": 20,
},
"offset": {
"type": "integer",
"description": "Offset for pagination",
"default": 0,
},
},
},
},
{
"name": "fireflies_get_transcript",
"description": "Get a single transcript with full text by ID",
"inputSchema": {
"type": "object",
"properties": {
"transcript_id": {
"type": "string",
"description": "The Fireflies transcript ID to retrieve",
},
},
"required": ["transcript_id"],
},
},
{
"name": "fireflies_search",
"description": "Search across all meeting transcripts by keyword",
"inputSchema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query string",
},
"limit": {
"type": "integer",
"description": "Maximum number of results",
"default": 20,
},
"offset": {
"type": "integer",
"description": "Offset for pagination",
"default": 0,
},
},
"required": ["query"],
},
},
{
"name": "fireflies_get_summary",
"description": "Get AI-generated meeting summary including topics, action items, questions",
"inputSchema": {
"type": "object",
"properties": {
"transcript_id": {
"type": "string",
"description": "The Fireflies transcript ID to summarize",
},
},
"required": ["transcript_id"],
},
},
{
"name": "fireflies_test_connection",
"description": "Validate the Fireflies API key and connectivity",
"inputSchema": {
"type": "object",
"properties": {},
},
},
]
def handle_tool_call(
self, tool_name: str, arguments: dict[str, Any]
) -> dict[str, Any]:
"""Route an MCP tool call to the appropriate handler.
Args:
tool_name: Name of the tool to invoke.
arguments: Tool arguments as a dictionary.
Returns:
Result dictionary. Contains an "error" key on failure.
"""
handlers = {
"fireflies_get_transcripts": self._get_transcripts,
"fireflies_get_transcript": self._get_transcript,
"fireflies_search": self._search,
"fireflies_get_summary": self._get_summary,
"fireflies_test_connection": self._test_connection,
}
handler = handlers.get(tool_name)
if handler is None:
return {"error": f"Unknown tool: {tool_name}"}
return handler(arguments)
@traces(spec="018", story="US-006", reqs=["REQ-018-066"])
def _get_transcripts(self, args: dict[str, Any]) -> dict[str, Any]:
"""Get list of recent transcripts.
Implements REQ-018-001 through REQ-018-005.
"""
return asyncio.run(
self._client.list_transcripts(
limit=args.get("limit", 20),
offset=args.get("offset", 0),
)
)
@traces(spec="018", story="US-006", reqs=["REQ-018-067"])
def _get_transcript(self, args: dict[str, Any]) -> dict[str, Any]:
"""Get single transcript with full text.
Implements REQ-018-006 through REQ-018-010.
"""
transcript_id = args.get("transcript_id")
return asyncio.run(
self._client.get_transcript(
transcript_id=transcript_id,
)
)
@traces(spec="018", story="US-006", reqs=["REQ-018-068"])
def _search(self, args: dict[str, Any]) -> dict[str, Any]:
"""Search across transcripts.
Implements REQ-018-016 through REQ-018-030.
"""
return asyncio.run(
self._client.search_transcripts(
query=args.get("query", ""),
limit=args.get("limit", 20),
offset=args.get("offset", 0),
)
)
@traces(spec="018", story="US-006", reqs=["REQ-018-069"])
def _get_summary(self, args: dict[str, Any]) -> dict[str, Any]:
"""Get AI-generated meeting summary.
Implements REQ-018-031 through REQ-018-040.
"""
transcript_id = args.get("transcript_id")
return asyncio.run(
self._client.summarize_transcript(
transcript_id=transcript_id,
)
)
@traces(
spec="018", story="US-006", reqs=["REQ-018-058", "REQ-018-072", "REQ-018-073"]
)
def _test_connection(self, args: dict[str, Any]) -> dict[str, Any]:
"""Validate Fireflies API connectivity.
Implements REQ-018-058 (connection validation via MCP bridge).
"""
return asyncio.run(self._client.test_connection())