initialize_data
Trigger GTFS data initialization to load bus schedules and routes for real-time tracking and arrival predictions.
Instructions
Manually trigger GTFS data initialization.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- src/catabus_mcp/server.py:196-205 (handler)Handler function for 'initialize_data' tool. Decorated with @mcp.tool (registration). Calls ensure_initialized() to load GTFS data if needed and returns initialization status.@mcp.tool async def initialize_data() -> Dict[str, Any]: """Manually trigger GTFS data initialization.""" await ensure_initialized() return { "status": "initialized" if initialized else "failed", "routes_loaded": len(gtfs_data.routes) if gtfs_data else 0, "stops_loaded": len(gtfs_data.stops) if gtfs_data else 0, "message": "GTFS data has been loaded" if initialized else "Failed to load GTFS data" }
- src/catabus_mcp/server.py:36-71 (helper)Supporting helper function called by initialize_data (and other tools) to perform lazy GTFS data loading with error handling and timeouts.async def ensure_initialized(): """Lazy initialization of GTFS data with comprehensive error handling.""" global gtfs_data, initialized if not initialized: logger.info("Starting GTFS data initialization...") # Initialize with empty data first to ensure server always works from .ingest.static_loader import GTFSData from .ingest.realtime_poll import RealtimeData try: # Try to load static GTFS data with aggressive timeout gtfs_data = await asyncio.wait_for( static_loader.load_feed(force_refresh=False, timeout_seconds=10), timeout=15.0 # Maximum 15 seconds for cloud environments ) logger.info(f"GTFS data loaded: {len(gtfs_data.routes)} routes, {len(gtfs_data.stops)} stops") except asyncio.TimeoutError: logger.warning("GTFS data loading timed out - using empty dataset") gtfs_data = GTFSData() except Exception as e: logger.warning(f"GTFS data loading failed: {e} - using empty dataset") gtfs_data = GTFSData() # Start realtime polling (non-blocking now) try: await realtime_poller.start() logger.info("Real-time polling started successfully") except Exception as e: logger.warning(f"Real-time polling failed to start: {e} - continuing without real-time data") realtime_poller.data = RealtimeData() initialized = True logger.info("Server initialization completed")