get_race_control_messages

Instructions

Implementation Reference

• tools/control/messages.py:10-92 (handler)

  Main handler function that fetches race control messages from FastF1, applies optional filtering by type, and returns structured Pydantic response.

  def get_race_control_messages(
      year: int,
      gp: Union[str, int],
      session: str,
      message_type: Optional[Literal["all", "penalties", "investigations", "flags", "safety_car"]] = "all"
  ) -> RaceControlMessagesResponse:
      """
      Get race control messages - flags, safety cars, investigations, penalties.
  
      Args:
          year: Season year (2018+)
          gp: Grand Prix name or round
          session: 'FP1', 'FP2', 'FP3', 'Q', 'S', 'R'
          message_type: Filter type - 'all', 'penalties', 'investigations', 'flags', 'safety_car' (default: 'all')
  
      Returns:
          RaceControlMessagesResponse with filtered messages
  
      Examples:
          get_race_control_messages(2024, "Monaco", "R") → All race control messages
          get_race_control_messages(2024, "Monaco", "R", "penalties") → Penalties only
          get_race_control_messages(2024, "Monaco", "R", "flags") → Flag periods only
      """
      session_obj = fastf1_client.get_session(year, gp, session)
      session_obj.load(laps=False, telemetry=False, weather=False, messages=True)
  
      event = session_obj.event
      messages_df = session_obj.race_control_messages
  
      # Define filter keywords based on message type
      filter_keywords = {
          "penalties": ['penalty', 'penalties', 'penalised', 'penalized', 'reprimand',
                        'time penalty', 'grid penalty', 'warning', 'fine', 'disqualified'],
          "investigations": ['investigation', 'investigated', 'under investigation', 'incident',
                            'noted', 'will be investigated', 'under review', 'reported'],
          "flags": ['yellow', 'red flag', 'green', 'double yellow', 'track clear', 'all clear'],
          "safety_car": ['safety car', 'virtual safety', 'vsc', 'sc deployed', 'sc ending']
      }
  
      # Convert to Pydantic models with optional filtering
      messages_list = []
      for idx, row in messages_df.iterrows():
          message_text = str(row.get('Message', '')).lower()
          category = str(row.get('Category', '')).lower()
          flag = str(row.get('Flag', '')).lower() if pd.notna(row.get('Flag')) else ''
  
          # Apply filter if not "all"
          if message_type != "all":
              keywords = filter_keywords.get(message_type, [])
  
              # Check if message matches filter
              matches = False
              if message_type == "flags":
                  # For flags, also check the Flag column
                  matches = (any(keyword in message_text for keyword in keywords) or
                            any(keyword in category for keyword in keywords) or
                            (pd.notna(row.get('Flag')) and row.get('Flag') not in ['', 'CLEAR']))
              else:
                  matches = (any(keyword in message_text for keyword in keywords) or
                            any(keyword in category for keyword in keywords))
  
              if not matches:
                  continue
  
          # Add message to list
          message = RaceControlMessage(
              time=str(row['Time']) if pd.notna(row.get('Time')) else None,
              category=str(row['Category']) if pd.notna(row.get('Category')) else None,
              message=str(row['Message']) if pd.notna(row.get('Message')) else None,
              status=str(row['Status']) if pd.notna(row.get('Status')) else None,
              flag=str(row['Flag']) if pd.notna(row.get('Flag')) else None,
              scope=str(row['Scope']) if pd.notna(row.get('Scope')) else None,
              sector=float(row['Sector']) if pd.notna(row.get('Sector')) else None,
              racing_number=str(row['RacingNumber']) if pd.notna(row.get('RacingNumber')) else None,
          )
          messages_list.append(message)
  
      return RaceControlMessagesResponse(
          session_name=session_obj.name,
          event_name=event['EventName'],
          messages=messages_list,
          total_messages=len(messages_list)
      )

• models/control/messages.py:5-25 (schema)

  Pydantic models defining the input/output schema for race control messages: RaceControlMessage (individual message) and RaceControlMessagesResponse (list wrapper with metadata).

  class RaceControlMessage(BaseModel):
      """Single race control message."""
  
      time: Optional[str] = Field(None, description="When the message was issued")
      category: Optional[str] = Field(None, description="Message category (e.g., Flag, SafetyCar, CarEvent)")
      message: Optional[str] = Field(None, description="The actual message text")
      status: Optional[str] = Field(None, description="Current session/track status")
      flag: Optional[str] = Field(None, description="Flag status (GREEN, YELLOW, RED, etc.)")
      scope: Optional[str] = Field(None, description="Scope of the message (Track, Sector, Driver)")
      sector: Optional[float] = Field(None, description="Relevant sector (if applicable)")
      racing_number: Optional[str] = Field(None, description="Driver number (if applicable)")
  
  
  class RaceControlMessagesResponse(BaseModel):
      """Race control messages response."""
  
      session_name: str = Field(description="Session name")
      event_name: str = Field(description="Grand Prix name")
      messages: list[RaceControlMessage] = Field(description="Race control messages")
      total_messages: int = Field(description="Total number of messages")

• server.py:164-164 (registration)

  MCP server tool registration decorator applied to the get_race_control_messages function.

  mcp.tool()(get_race_control_messages)