get_telemetry

Instructions

Implementation Reference

• src/f1_mcp_server/f1_data.py:443-519 (handler)

  The actual implementation of get_telemetry: calls fastf1 to load session/laps, picks the driver's specific lap number or fastest lap, retrieves telemetry data, serializes it with json_serial, and returns status+data

  def get_telemetry(
      year, event_identifier, session_name, driver_identifier, lap_number=None
  ):
      """
      Get telemetry data for a specific lap or fastest lap.
  
      Args:
          year (int or str): The year of the F1 season
          event_identifier (str): Event name or round number
          session_name (str): Session type (Race, Qualifying, Sprint, etc.)
          driver_identifier (str): Driver number, code, or name
          lap_number (int, optional): Specific lap number or None for fastest lap
  
      Returns:
          dict: Status and telemetry data or error information
      """
      try:
          year = int(year)
          session = fastf1.get_session(year, event_identifier, session_name)
          session.load()
  
          # Get laps for the specified driver
          driver_laps = session.laps.pick_driver(driver_identifier)
  
          if len(driver_laps) == 0:
              return {
                  "status": "error",
                  "message": f"No laps found for driver {driver_identifier}",
              }
  
          # Get the specific lap or fastest lap
          if lap_number:
              matching_laps = driver_laps[driver_laps["LapNumber"] == int(lap_number)]
              if len(matching_laps) == 0:
                  return {
                      "status": "error",
                      "message": (
                          f"Lap number {lap_number} not found for driver "
                          f"{driver_identifier}"
                      ),
                  }
              lap = matching_laps.iloc[0]
          else:
              lap = driver_laps.pick_fastest()
              if lap is None:
                  return {
                      "status": "error",
                      "message": "No valid fastest lap found for driver "
                      f"{driver_identifier}",
                  }
  
          # Get telemetry data
          telemetry = lap.get_telemetry()
  
          # Convert to JSON serializable format
          telemetry_dict = telemetry.to_dict(orient="records")
          clean_data = []
  
          for item in telemetry_dict:
              clean_item = {k: json_serial(v) for k, v in item.items()}
              clean_data.append(clean_item)
  
          # Add lap information
          lap_info = {
              "LapNumber": int(lap["LapNumber"])
              if not pd.isna(lap["LapNumber"])
              else None,
              "LapTime": str(lap["LapTime"]) if not pd.isna(lap["LapTime"]) else None,
              "Compound": lap["Compound"] if not pd.isna(lap["Compound"]) else None,
              "TyreLife": int(lap["TyreLife"]) if not pd.isna(lap["TyreLife"]) else None,
          }
  
          result = {"lapInfo": lap_info, "telemetry": clean_data}
  
          return {"status": "success", "data": result}
      except Exception as e:
          return {"status": "error", "message": str(e)}

• src/f1_mcp_server/server.py:466-511 (schema)

  Input schema registration for the 'get_telemetry' tool: defines properties year (number), event_identifier (string), session_name (string), driver_identifier (string), lap_number (optional number)

  types.Tool(
      name="get_telemetry",
      description=("Get telemetry data for a specific Formula One lap"),
      inputSchema={
          "type": "object",
          "properties": {
              "year": {
                  "type": "number",
                  "description": "Season year (e.g., 2023)",
              },
              "event_identifier": {
                  "type": "string",
                  "description": (
                      "Event name or round number (e.g., 'Monaco' or '7')"
                  ),
              },
              "session_name": {
                  "type": "string",
                  "description": (
                      "Session name (e.g., 'Race', 'Qualifying', "
                      "'Sprint', 'FP1', 'FP2', 'FP3')"
                  ),
              },
              "driver_identifier": {
                  "type": "string",
                  "description": (
                      "Driver identifier (number, code, or name; "
                      "e.g., '44', 'HAM', 'Hamilton')"
                  ),
              },
              "lap_number": {
                  "type": "number",
                  "description": (
                      "Lap number (optional, gets fastest lap if not "
                      "provided)"
                  ),
              },
          },
          "required": [
              "year",
              "event_identifier",
              "session_name",
              "driver_identifier",
          ],
      },
  ),

• src/f1_mcp_server/server.py:224-240 (registration)

  Handler dispatch block: routes the name 'get_telemetry' to the get_telemetry function, validates lap_number, and passes sanitized arguments

  elif name == "get_telemetry":
      lap_number = arguments.get("lap_number")
      if lap_number is not None:
          try:
              lap_number = int(lap_number)
              if lap_number <= 0:
                  raise ValueError("Lap number must be positive")
          except (ValueError, TypeError) as e:
              raise ValueError(f"Invalid lap number: {lap_number}") from e
  
      result = get_telemetry(
          sanitized_args["year"],
          str(arguments["event_identifier"]),
          str(arguments["session_name"]),
          str(arguments["driver_identifier"]),
          lap_number,
      )

• src/f1_mcp_server/server.py:32-32 (registration)

  Import of get_telemetry from f1_data module into server.py

  get_telemetry,

• src/f1_mcp_server/f1_data.py:45-63 (helper)

  The json_serial helper used by get_telemetry to convert non-JSON serializable objects (timestamps, numpy types, NaN) to JSON-safe values

  def json_serial(obj: Any) -> str | int | float | None:
      """
      Convert non-JSON serializable objects to strings.
  
      Args:
          obj: Object to be serialized to JSON
  
      Returns:
          JSON serializable representation of the object
      """
      if isinstance(obj, datetime | pd.Timestamp):
          return obj.isoformat()
      if isinstance(obj, np.integer):
          return int(obj)
      if isinstance(obj, np.floating):
          return float(obj)
      if pd.isna(obj) or obj is None:
          return None
      return str(obj)