Caltrain MCP Server

Instructions

Input Schema

Input Schema (JSON Schema)

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()