CATA Bus MCP Server

Instructions

Implementation Reference

• src/catabus_mcp/server.py:146-158 (handler)

  Primary handler function for the 'trip_alerts_tool' MCP tool. Registered via @mcp.tool decorator. Ensures GTFS data initialization and calls the helper function to retrieve filtered alerts.

  @mcp.tool
  async def trip_alerts_tool(route_id: str | None = None) -> list[dict[str, Any]]:
      """Get current service alerts.
  
      Args:
          route_id: Optional route ID to filter alerts (if None, returns all)
  
      Returns:
          List of alerts with route ID, header, description, and severity
      """
      await ensure_initialized()
      return await trip_alerts(realtime_poller.data, route_id)

• src/catabus_mcp/tools/trip_alerts.py:8-51 (helper)

  Core implementation logic for extracting, filtering (by route_id), and formatting service alerts from GTFS realtime data's alerts list.

  async def trip_alerts(
      realtime_data: RealtimeData, route_id: str | None = None
  ) -> list[dict[str, Any]]:
      """
      Get current service alerts.
  
      Args:
          realtime_data: The GTFS realtime data.
          route_id: Optional route ID to filter alerts (if None, returns all).
  
      Returns:
          List of alerts with route ID, header, description, and severity.
      """
      alerts = []
  
      for alert in realtime_data.alerts:
          # Check if this alert is relevant to the requested route
          if route_id:
              relevant = False
              for entity in alert.informed_entities:
                  if entity.get("route_id") == route_id:
                      relevant = True
                      break
              if not relevant:
                  continue
  
          # Get affected route IDs
          affected_routes = []
          for entity in alert.informed_entities:
              if "route_id" in entity:
                  affected_routes.append(entity["route_id"])
  
          alerts.append(
              {
                  "route_id": affected_routes[0] if len(affected_routes) == 1 else None,
                  "affected_routes": affected_routes,
                  "header": alert.header,
                  "description": alert.description,
                  "severity": alert.severity,
              }
          )
  
      return alerts