Caltrain MCP Server

Instructions

Implementation Reference

• src/caltrain_mcp/server.py:16-131 (handler)

  The primary handler for the 'next_trains' MCP tool. It is registered via the @mcp.tool() decorator, parses inputs, handles station name resolution with fuzzy matching and error messages, retrieves GTFS data, finds the next trains using the gtfs helper, and formats the response.

  @mcp.tool()
  async def next_trains(
      origin: str, destination: str, when_iso: str | None = None
  ) -> str:
      """Return the next few scheduled Caltrain departures.
  
      Args:
          origin: Station name (e.g. 'San Jose Diridon', 'Palo Alto', 'San Francisco').
                  Supports common abbreviations like 'SF' for San Francisco, 'SJ' for San Jose.
                  If station is not found, use list_stations() to see all available options.
          destination: Station name (e.g. 'San Francisco', 'Mountain View', 'Tamien').
                       Supports common abbreviations like 'SF' for San Francisco, 'SJ' for San Jose.
                       If station is not found, use list_stations() to see all available options.
          when_iso: Optional ISO-8601 datetime (local time). Default: now.
  
      Note: If you get a "Station not found" error, try using the list_stations() tool first
      to see exact station names, then retry with the correct spelling.
      """
      try:
          # Parse the target time
          if when_iso:
              try:
                  when_dt = datetime.fromisoformat(when_iso.replace("Z", "+00:00"))
                  # Convert to local time if needed
                  if when_dt.tzinfo is not None:
                      # Convert to naive datetime assuming Pacific time
                      when_dt = when_dt.replace(tzinfo=None)
              except ValueError:
                  return (
                      f"Invalid datetime format: {when_iso}. Please use ISO-8601 format."
                  )
          else:
              when_dt = datetime.now()
  
          target_date = when_dt.date()
          seconds_since_midnight = (
              when_dt.hour * 3600 + when_dt.minute * 60 + when_dt.second
          )
  
          # Find station IDs with enhanced error handling
          data = gtfs.get_default_data()
          try:
              origin_id = gtfs.find_station(origin, data)
          except ValueError:
              available_stations = gtfs.list_all_stations(data)
              # Try to find close matches
              close_matches = [
                  s
                  for s in available_stations
                  if origin.lower() in s.lower()
                  or s.lower().startswith(origin.lower()[:3])
              ]
              error_msg = f"Origin station '{origin}' not found."
              if close_matches:
                  error_msg += (
                      f" Did you mean one of these? {', '.join(close_matches[:5])}"
                  )
              else:
                  error_msg += " Use list_stations() to see all available stations."
              return error_msg
  
          try:
              dest_id = gtfs.find_station(destination, data)
          except ValueError:
              available_stations = gtfs.list_all_stations(data)
              # Try to find close matches
              close_matches = [
                  s
                  for s in available_stations
                  if destination.lower() in s.lower()
                  or s.lower().startswith(destination.lower()[:3])
              ]
              error_msg = f"Destination station '{destination}' not found."
              if close_matches:
                  error_msg += (
                      f" Did you mean one of these? {', '.join(close_matches[:5])}"
                  )
              else:
                  error_msg += " Use list_stations() to see all available stations."
              return error_msg
  
          # Get station names for display
          origin_name = gtfs.get_station_name(origin_id, data)
          dest_name = gtfs.get_station_name(dest_id, data)
  
          # Find next trains
          trains = gtfs.find_next_trains(
              origin_id,
              dest_id,
              seconds_since_midnight,
              target_date,
              data,
          )
  
          if not trains:
              return f"No more trains today from {origin_name} to {dest_name}."
  
          # Format results
          lines = []
          for dep_time, arr_time, train_name, headsign in trains:
              line = f"• Train {train_name}: {dep_time} → {arr_time}"
              if headsign:
                  line += f" (to {headsign})"
              lines.append(line)
  
          date_str = target_date.strftime("%A, %B %d, %Y")
          current_time_str = when_dt.strftime("%I:%M %p")
          header = (
              f"Next Caltrain departures from {origin_name} to {dest_name} "
              f"on {date_str}:\n(Current time: {current_time_str})\n\n"
          )
          return header + "\n".join(lines)
  
      except Exception as e:
          return f"Error: {str(e)}"

• src/caltrain_mcp/gtfs.py:230-311 (helper)

  Core helper function that performs the GTFS query to find next trains between stations on a given date after a specific time. Filters active services, joins stop times, ensures directionality, and returns formatted train details. Called by the next_trains handler.

  def find_next_trains(
      origin_station_id: str,
      destination_station_id: str,
      after_seconds: int,
      target_date: date,
      data: GTFSData,
      limit: int = 5,
  ) -> list[tuple[str, str, str, str]]:
      """Find the next trains from origin to destination."""
  
      trips_df = data.trips
      stop_times_df = data.stop_times
  
      # Get active service IDs for the target date
      service_ids = get_active_service_ids(target_date, data)
      if not service_ids:
          return []
  
      # Filter trips to only those running today
      active_trips = trips_df[trips_df["service_id"].isin(service_ids)]
  
      if active_trips.empty:
          return []
  
      # Get platform stops for both stations
      origin_platforms = get_platform_stops_for_station(origin_station_id, data)
      dest_platforms = get_platform_stops_for_station(destination_station_id, data)
  
      if not origin_platforms or not dest_platforms:
          return []
  
      # Get stop times for origin platforms
      origin_times = stop_times_df[stop_times_df["stop_id"].isin(origin_platforms)].copy()
  
      # Get stop times for destination platforms
      dest_times = stop_times_df[stop_times_df["stop_id"].isin(dest_platforms)].copy()
  
      # Join on trip_id to get trips that serve both stations
      combined = origin_times.merge(
          dest_times, on="trip_id", suffixes=("_origin", "_dest")
      )
  
      # Only keep trips where destination comes after origin (higher stop_sequence)
      combined = combined[
          combined["stop_sequence_dest"] > combined["stop_sequence_origin"]
      ]
  
      # Only keep trips that are active today
      combined = combined[combined["trip_id"].isin(active_trips["trip_id"])]
  
      if combined.empty:
          return []
  
      # Convert departure times to seconds
      combined["dep_seconds"] = combined["departure_time_origin"].apply(time_to_seconds)
      combined = combined.dropna(subset=["dep_seconds"])
  
      # Filter to departures after the specified time
      upcoming = combined[combined["dep_seconds"] >= after_seconds]
  
      if upcoming.empty:
          return []
  
      # Sort by departure time and limit results
      upcoming = upcoming.sort_values("dep_seconds").head(limit)
  
      # Join with trips to get trip information
      upcoming = upcoming.merge(
          active_trips[["trip_id", "trip_headsign", "trip_short_name"]], on="trip_id"
      )
  
      results = []
      for _, row in upcoming.iterrows():
          dep_time = row["departure_time_origin"]
          arr_time = row["arrival_time_dest"]
          train_name = row["trip_short_name"] or row["trip_id"]
          headsign = row["trip_headsign"] or ""
  
          results.append((dep_time, arr_time, train_name, headsign))
  
      return results

• src/caltrain_mcp/server.py:17-33 (schema)

  Input schema defined by function parameters (origin: str, destination: str, when_iso: str | None) and comprehensive docstring explaining usage, abbreviations, and error handling. Output is a formatted string list of trains.

  async def next_trains(
      origin: str, destination: str, when_iso: str | None = None
  ) -> str:
      """Return the next few scheduled Caltrain departures.
  
      Args:
          origin: Station name (e.g. 'San Jose Diridon', 'Palo Alto', 'San Francisco').
                  Supports common abbreviations like 'SF' for San Francisco, 'SJ' for San Jose.
                  If station is not found, use list_stations() to see all available options.
          destination: Station name (e.g. 'San Francisco', 'Mountain View', 'Tamien').
                       Supports common abbreviations like 'SF' for San Francisco, 'SJ' for San Jose.
                       If station is not found, use list_stations() to see all available options.
          when_iso: Optional ISO-8601 datetime (local time). Default: now.
  
      Note: If you get a "Station not found" error, try using the list_stations() tool first
      to see exact station names, then retry with the correct spelling.
      """

• src/caltrain_mcp/server.py:16-16 (registration)

  The @mcp.tool() decorator registers the next_trains function as an MCP tool with FastMCP instance.

  @mcp.tool()