Fathom-Simple-MCP

Instructions

Implementation Reference

• tools/search.py:135-257 (handler)

  The core handler function implementing the search_meetings tool logic. It normalizes the query, fetches meetings with pagination, optionally retrieves transcripts, filters matches using helper functions across metadata and transcripts, formats results, and returns structured output.

  async def search_meetings(
      ctx: Context,
      query: str,
      include_transcript: bool = False
  ) -> dict:
      """Search meetings by keyword across titles, participants, teams, topics, and optionally transcripts.
      
      This tool searches meeting metadata and optionally full transcript content, returning
      matching meetings with their recording_id, summary, and optionally transcripts. Uses fuzzy matching
      to handle partial matches, plurals, and case-insensitive search.
      
      Fetches up to 10 pages (500 meetings max) to provide comprehensive search results.
      
      Args:
          ctx: MCP context for logging
          query: Search query string to match against meeting metadata
          include_transcript: If True, search within transcripts and include them in results (default: False)
      
      Returns:
          dict: {
              "items": [Meeting objects matching the search query with summary and optionally transcripts],
              "query": str (the search query used),
              "total_matches": int (number of matches found),
              "searched_transcripts": bool (whether transcripts were searched)
          }
      """
      try:
          await ctx.info(f"Searching meetings with query: {query}")
          
          if not query or not query.strip():
              await ctx.error("Search query cannot be empty")
              return {
                  "items": [],
                  "query": query,
                  "total_matches": 0
              }
          
          # Normalize the search query
          search_normalized = _normalize_search(query)
          
          # Fetch all meetings (with pagination, max 10 pages = 500 meetings)
          all_meetings = []
          cursor: Optional[str] = None
          page = 1
          max_pages = 10
          
          while page <= max_pages:
              await ctx.info(f"Fetching meetings page {page}/{max_pages}")
              
              params = {
                  "include_summary": True  # Always include summaries in search results
              }
              if cursor:
                  params["cursor"] = cursor
              
              response = await client.get_meetings(params=params)
              items = response.get("items", [])
              
              if not items:
                  break
              
              all_meetings.extend(items)
              
              # Check for next page
              cursor = response.get("cursor")
              if not cursor:
                  break
              
              page += 1
          
          await ctx.info(f"Total meetings fetched: {len(all_meetings)}")
          
          # If including transcripts, fetch them for meetings that don't have them
          if include_transcript:
              await ctx.info("Fetching transcripts for meetings...")
              for meeting in all_meetings:
                  if not meeting.get("transcript"):
                      recording_id = meeting.get("recording_id")
                      if recording_id:
                          try:
                              transcript_data = await client.get_transcript(recording_id)
                              meeting["transcript"] = transcript_data.get("transcript")
                          except Exception as e:
                              await ctx.info(f"Could not fetch transcript for {recording_id}: {str(e)}")
          
          # Filter meetings by search query and track where matches are found
          matched_meetings = []
          
          if include_transcript:
              for m in all_meetings:
                  matches, found_in_transcript = _meeting_matches_search_with_transcript(m, search_normalized)
                  if matches:
                      matched_meetings.append((m, found_in_transcript))
          else:
              for m in all_meetings:
                  matches, _ = _meeting_matches_search(m, search_normalized)
                  if matches:
                      matched_meetings.append((m, False))
          
          # Apply field filtering
          filtered_meetings = [
              _filter_meeting_fields(m, found_in_transcript=found_in_transcript)
              for m, found_in_transcript in matched_meetings
          ]
          
          await ctx.info(
              f"Search completed: found {len(matched_meetings)} matches out of {len(all_meetings)} meetings"
          )
          
          return {
              "items": filtered_meetings,
              "query": query,
              "total_matches": len(matched_meetings),
              "searched_transcripts": include_transcript
          }
          
      except FathomAPIError as e:
          await ctx.error(f"Fathom API error during search: {e.message}")
          raise e
      except Exception as e:
          await ctx.error(f"Unexpected error during search: {str(e)}")
          raise e

• server.py:76-104 (registration)

  The MCP tool registration using @mcp.tool decorator. Defines the input schema with Pydantic Field descriptions and docstring, and delegates execution to the implementation in tools/search.py.

  @mcp.tool
  async def search_meetings(
      ctx: Context,
      query: str = Field(
          ...,
          description="Search query to match against meeting metadata (titles, participants, teams, topics, summaries, and optionally transcripts)",
      ),
      include_transcript: bool = Field(
          default=False,
          description="If True, search within transcripts and include them in results.",
      ),
  ) -> Dict[str, Any]:
      """Search meetings by keyword across metadata fields and optionally transcripts.
  
      This tool searches meeting metadata (titles, attendees, teams, topics, summaries) and optionally
      full transcript content. Uses fuzzy matching to handle partial matches, plurals, and case-insensitive search.
  
      By default, transcripts are NOT searched or included to optimize performance. Set include_transcript=True
      to search within and return transcript data.
  
      Fetches all meetings (with pagination) and returns those matching the search query.
  
      Examples:
          search_meetings(\"McDonalds\")  # Search metadata only
          search_meetings(\"budget discussion\", include_transcript=True)  # Search including transcripts
          search_meetings(\"engineering\")  # Find meetings related to engineering
      """
      return await tools.search.search_meetings(ctx, query, include_transcript)

• tools/search.py:57-106 (helper)

  Helper function to check if a meeting matches the search query in metadata fields like title, participants, teams, topics, and summary.

  def _meeting_matches_search(meeting: dict, search_normalized: str) -> tuple:
      """Check if a meeting matches the search term in title, attendees, or teams.
      
      Returns:
          tuple: (matches, found_in_transcript) - matches=True if found, found_in_transcript=False for metadata searches
      """
      
      # Check title
      title = _normalize_search(meeting.get("title") or "")
      if search_normalized in title:
          return True, False
  
      # Check meeting_title
      meeting_title = _normalize_search(meeting.get("meeting_title") or "")
      if search_normalized in meeting_title:
          return True, False
  
      # Check attendee names and emails
      for invitee in meeting.get("calendar_invitees") or []:
          name = _normalize_search(invitee.get("name") or "")
          email = (invitee.get("email") or "").lower()
          if search_normalized in name or search_normalized in email:
              return True, False
  
      # Check team names
      for team in meeting.get("teams") or []:
          team_name = _normalize_search(team.get("name") or "") if isinstance(team, dict) else _normalize_search(str(team))
          if search_normalized in team_name:
              return True, False
  
      # Check topics
      for topic in meeting.get("topics") or []:
          topic_text = _normalize_search(topic.get("name") or "") if isinstance(topic, dict) else _normalize_search(str(topic))
          if search_normalized in topic_text:
              return True, False
  
      # Check default_summary.markdown_formatted (summary)
      summary = meeting.get("default_summary")
      summary_text = None
  
      if isinstance(summary, dict):
          summary_text = summary.get("markdown_formatted")
  
      if summary_text:
          summary_text_norm = _normalize_search(str(summary_text))
          if search_normalized in summary_text_norm:
              return True, False
  
      return False, False

• tools/search.py:15-55 (helper)

  Helper function to filter and structure the meeting data returned in search results, extracting relevant fields including summary.

  def _filter_meeting_fields(meeting: dict, found_in_transcript: bool = False) -> dict:
      """Filter and structure meeting fields for search results.
      
      Args:
          meeting: The meeting object from the API
          found_in_transcript: Whether the search match was found in the transcript
      
      Returns:
          dict: Filtered meeting fields with summary and optional found_in_transcript flag
      """
      summary = meeting.get("default_summary")
      summary_text = None
      if isinstance(summary, dict):
          summary_text = summary.get("markdown_formatted")
      elif isinstance(summary, str):
          summary_text = summary
      
      result = {
          "title": meeting.get("title"),
          "recording_id": meeting.get("recording_id"),
          "url": meeting.get("url"),
          "share_url": meeting.get("share_url"),
          "created_at": meeting.get("created_at"),
          "scheduled_start_time": meeting.get("scheduled_start_time"),
          "scheduled_end_time": meeting.get("scheduled_end_time"),
          "recording_start_time": meeting.get("recording_start_time"),
          "recording_end_time": meeting.get("recording_end_time"),
          "transcript_language": meeting.get("transcript_language"),
          "calendar_invitees": meeting.get("calendar_invitees"),
          "recorded_by": meeting.get("recorded_by"),
          "teams": meeting.get("teams"),
          "topics": meeting.get("topics"),
          "summary": summary_text
      }
      
      # Add flag indicating if match was found in transcript
      if found_in_transcript:
          result["found_in_transcript"] = True
      
      return result

• tools/search.py:6-12 (helper)

  Helper function for normalizing search text: lowercases, removes spaces/hyphens/underscores, strips trailing 's' for plural handling.

  def _normalize_search(text: str) -> str:
      """Normalize text for fuzzy matching: lowercase, remove spaces/hyphens, strip trailing 's'."""
      normalized = text.lower().replace(" ", "").replace("-", "").replace("_", "")
      # Strip trailing 's' to handle simple plurals (labs -> lab, meetings -> meeting)
      if normalized.endswith("s") and len(normalized) > 2:
          normalized = normalized[:-1]
      return normalized