Nagoya Bus MCP

Instructions

Implementation Reference

• nagoya_bus_mcp/mcp/server.py:40-40 (registration)

  Registration of the get_timetable tool with the FastMCP server.

  mcp_server.tool(get_timetable)

• nagoya_bus_mcp/mcp/tools.py:21-34 (schema)

  Pydantic schemas for the timetable data structures used in get_timetable response.

  class TimeTable(BaseModel):
      route: Annotated[str, Field(description="路線")]
      direction: Annotated[str, Field(description="方面")]
      pole: Annotated[str, Field(description="乗り場")]
      stop_stations: Annotated[list[str], Field(description="停車バス停のリスト")]
      timetable: Annotated[dict[str, list[str]], Field(description="曜日別の時刻表")]
      url: str
  
  
  class TimeTableResponse(BaseModel):
      station_number: int
      timetables: list[TimeTable]
      url: str

• nagoya_bus_mcp/mcp/tools.py:96-138 (handler)

  The core handler function that implements the get_timetable tool logic: retrieves station diagram from client, processes it into TimeTable entries, and returns TimeTableResponse.

  async def get_timetable(ctx: Context, station_number: int) -> TimeTableResponse | None:
      """Get timetable for a given station."""
      client = ctx.request_context.lifespan_context.bus_client
  
      station_name = (await _get_station_numbers(client)).get(station_number)
      if station_name is None:
          log.warning("Station number %s not found", station_number)
          return None
  
      log.info("Getting timetable for station number %s", station_number)
      diagram_response = await client.get_station_diagram(station_number)
      if not diagram_response.root:
          log.error(
              "Failed to get station diagram for %s (%s)", station_name, station_number
          )
          return None
  
      timetables: list[TimeTable] = []
      for line, railways in diagram_response.root.items():
          for railway_index, railway in enumerate(railways):
              diagram: dict[str, list[str]] = {}
              for day, hour_minutes in railway.diagram.root.items():
                  diagram[day] = []
                  for hour, minutes in hour_minutes.items():
                      for minute in minutes:
                          diagram[day].append(f"{hour}:{minute:02}")
              timetables.append(
                  TimeTable(
                      route=line,
                      direction="・".join(railway.railway),
                      stop_stations=reduce(iadd, railway.stations, []),
                      pole=railway.polename,
                      timetable=diagram,
                      url=f"{client.base_url}/jp/pc/bus/timetable_dtl.html?name={station_name}&keito={line}&lineindex={railway_index}",
                  )
              )
  
      return TimeTableResponse(
          timetables=timetables,
          station_number=station_number,
          url=f"{client.base_url}/jp/pc/bus/timetable_list.html?name={station_name}&toname=",
      )