KMB Bus MCP Server

Instructions

Args:
    route: The bus route number (e.g., "1A", "6", "960")
    stop_name: The name of the bus stop

Implementation Reference

• kmb_mcp.py:136-188 (handler)

  The handler function decorated with @mcp.tool(), implementing the core logic for retrieving and formatting the next bus arrival times for a given route and stop name. Includes input schema via type hints and docstring Args.

  @mcp.tool()
  async def get_next_bus(route: str, stop_name: str) -> str:
      """Get the next arrival time for a specified bus route at a stop.
      
      Args:
          route: The bus route number (e.g., "1A", "6", "960")
          stop_name: The name of the bus stop
      """
      # Find the stop ID by name
      stops = await find_stops_by_name(stop_name)
      
      if not stops:
          return f"Could not find any stops matching '{stop_name}'"
      
      results = []
      for stop in stops:
          stop_id = stop["stop"]
          stop_name_en = stop["name_en"]
          
          # Get ETA data
          eta_data = await get_eta(stop_id, route)
  
          
          if not eta_data:
              results.append(f"No arrival data available for route {route} at stop '{stop_name_en}' ({stop_id})")
              continue
          
          # Filter ETAs for the specified route
          route_etas = [eta for eta in eta_data if eta["route"] == route]
          
          if not route_etas:
              results.append(f"No scheduled arrivals for route {route} at stop '{stop_name_en}' ({stop_id})")
              continue
          
          # Format ETA information
          stop_results = [f"Arrivals for route {route} at '{stop_name_en}' ({stop_id}):"]
          
          for eta in route_etas:
              eta_time = eta.get("eta", None)
              if eta_time :
                  eta_time = eta_time.split("+")[0].replace("T", " ")  # Format the timestamp
              
              dest = eta.get("dest_tc", "") or eta.get("dest_en", "Unknown destination")
              remark = eta.get("rmk_tc", "") or eta.get("rmk_en", "")
              
              if remark:
                  stop_results.append(f"- {eta_time} to {dest} ({remark})")
              else:
                  stop_results.append(f"- {eta_time} to {dest}")
          
          results.append("\n".join(stop_results))
      
      return "\n\n".join(results)