server.py

"""
MCP Server using FastMCP that adjusts pipette settings based on liquid type and environment
"""

from mcp.server.fastmcp import FastMCP
from typing import Optional
import os
import json
import glob
import math
import re

# Initialize the FastMCP server
mcp = FastMCP(
    name="LiquidLib",
    description="This MCP server provides pipette dispense speed recommendations based on liquid type, temperature, and relative humidity."
)

# User-specific label (optional)
USER_NAME = "LiquidLab User"

# Directory where the liquid JSON files are stored
LIQUIDS_DIR = "liquids"

# @mcp.tool()
# def adjust_dispense_volume(
#     temperature_c: float,
#     humidity_percent: float,
#     liquid_name: str
# ) -> str:
#     """
#     Adjusts pipette dispense volume based on temperature (°C), humidity (% RH), and liquid type.
#     Extracts density from structured PubChem-style JSON if available.
#     """

#     # === Input validation ===
#     if not (isinstance(temperature_c, (int, float)) and math.isfinite(temperature_c)):
#         return "Error: Temperature must be a finite number between 0 and 100 °C."
#     if not (isinstance(humidity_percent, (int, float)) and math.isfinite(humidity_percent)):
#         return "Error: Relative humidity must be a finite number between 0% and 100%."
#     if not (0 <= temperature_c <= 32.2):
#         return "Error: Temperature must be between 10 and 32.2 °C."
#     if not (0 <= humidity_percent <= 100):
#         return "Error: Relative humidity must be between 0% and 100%."

#     # === File matching ===
#     liquid_name = liquid_name.strip().lower()
#     pattern = os.path.join(LIQUIDS_DIR, f"*{liquid_name}*.json")
#     matching_files = glob.glob(pattern)

#     if not matching_files:
#         return f"Error: No data file found for liquid containing '{liquid_name}'. (Pattern: {pattern})"

#     if len(matching_files) > 1:
#         matched = ", ".join(os.path.basename(f) for f in matching_files)
#         return f"Error: Multiple files matched for '{liquid_name}': {matched}. Please be more specific."

#     filename = matching_files[0]

#     # === Parse structured JSON ===
#     try:
#         with open(filename, 'r') as f:
#             data = json.load(f)

#         if not isinstance(data, list):
#             return f"Error: Expected a list of sections in the JSON structure."

#         density_val = None

#         # Look for "Density" section
#         for section in data:
#             if section.get("TOCHeading", "").lower() == "density":
#                 for info in section.get("Information", []):
#                     val_obj = info.get("Value", {})
#                     for string_entry in val_obj.get("StringWithMarkup", []):
#                         s = string_entry.get("String", "")
#                         match = re.search(r"([0-9.]+)\s*g/(?:cu\s*cm|cm3|cc|mL)", s, re.IGNORECASE)
#                         if match:
#                             density_val = float(match.group(1))
#                             break
#                     if density_val is not None:
#                         break
#             if density_val is not None:
#                 break

#         if density_val is None:
#             return f"Error: Could not extract numeric density from '{os.path.basename(filename)}'."

#         # Estimate pipette base volume from density
#         base = density_val * 1000  # µL (arbitrary scaling factor)
#         temp_coeff = 0.01
#         hum_coeff = -0.002

#     except Exception as e:
#         return f"Error reading or parsing JSON file '{os.path.basename(filename)}': {e}"

#     # === Volume adjustment ===
#     adjusted_volume = base + (temperature_c * temp_coeff) + (humidity_percent * hum_coeff)

#     return (
#         f"Matched file: {os.path.basename(filename)}\n"
#         f"Extracted density: {density_val:.4f} g/cm³\n"
#         f"Base volume estimate: {base:.2f} µL\n"
#         f"Adjusted for {temperature_c}°C and {humidity_percent}% RH: {adjusted_volume:.2f} µL"
#     )

@mcp.tool()
def adjust_dispense_speed(
    temperature_c: float,
    humidity_percent: float,
    liquid_name: str
) -> str:
    """
    Adjusts pipette dispense speed (µL/s) based on temperature (°C), humidity (% RH), and liquid type.
    Extracts density from structured JSON. Higher density → slower speed. Higher temp → faster speed.
    """

    # === Input validation ===
    if not (isinstance(temperature_c, (int, float)) and math.isfinite(temperature_c)):
        return "Error: Temperature must be a finite number between 0 and 32.2 °C."
    if not (isinstance(humidity_percent, (int, float)) and math.isfinite(humidity_percent)):
        return "Error: Relative humidity must be a finite number between 0% and 100%."
    if not (0 <= temperature_c <= 32.2):
        return "Error: Temperature must be between 0 and 32.2 °C."
    if not (0 <= humidity_percent <= 100):
        return "Error: Relative humidity must be between 0% and 100%."

    # === File matching ===
    liquid_name = liquid_name.strip().lower()
    pattern = os.path.join(LIQUIDS_DIR, f"*{liquid_name}*.json")
    matching_files = glob.glob(pattern)

    if not matching_files:
        return f"Error: No data file found for liquid containing '{liquid_name}'. (Pattern: {pattern})"

    if len(matching_files) > 1:
        matched = ", ".join(os.path.basename(f) for f in matching_files)
        return f"Error: Multiple files matched for '{liquid_name}': {matched}. Please be more specific."

    filename = matching_files[0]

    # === Parse structured JSON ===
    try:
        with open(filename, 'r') as f:
            data = json.load(f)

        if not isinstance(data, list):
            return f"Error: Expected a list of sections in the JSON structure."

        density_val = None

        for section in data:
            if section.get("TOCHeading", "").lower() == "density":
                for info in section.get("Information", []):
                    val_obj = info.get("Value", {})
                    for string_entry in val_obj.get("StringWithMarkup", []):
                        s = string_entry.get("String", "")
                        match = re.search(r"([0-9.]+)\s*g/(?:cu\s*cm|cm3|cc|mL)", s, re.IGNORECASE)
                        if match:
                            density_val = float(match.group(1))
                            break
                    if density_val is not None:
                        break
            if density_val is not None:
                break

        if density_val is None:
            return f"Error: Could not extract numeric density from '{os.path.basename(filename)}'."

    except Exception as e:
        return f"Error reading or parsing JSON file '{os.path.basename(filename)}': {e}"

    # === Dispense speed model ===
    base_speed = 50.0  # µL/s default
    # Higher density → lower speed; higher temp → higher speed
    # Simple model: inverse proportional to density, direct to temp
    speed_factor = (temperature_c + 1) / density_val  # Avoid division by zero
    adjusted_speed = base_speed * speed_factor

    return (
        f"Matched file: {os.path.basename(filename)}\n"
        f"Extracted density: {density_val:.4f} g/cm³\n"
        f"Base speed: {base_speed:.2f} µL/s\n"
        f"Adjusted speed for {temperature_c}°C and {humidity_percent}% RH: {adjusted_speed:.2f} µL/s"
    )


if __name__ == "__main__":
    # Run the server
    mcp.run(transport='stdio')