mcp-stargazing

Instructions

Implementation Reference

• src/functions/celestial/impl.py:123-145 (handler)

  The primary handler for the 'get_constellation' MCP tool. Decorated with @mcp.tool() for registration. Processes input parameters, calls helper function get_constellation_center in a thread, and returns formatted response. Function parameters and docstring define the input schema.
  @mcp.tool() async def get_constellation( constellation_name: str, lon: float, lat: float, time: str, time_zone: str ) -> Dict[str, Any]: """Get the position (altitude/azimuth) of the center of a constellation. Args: constellation_name: Name of constellation (e.g. "Orion", "Ursa Major") lon: Observer longitude in degrees lat: Observer latitude in degrees time: Observation time string "YYYY-MM-DD HH:MM:SS" time_zone: IANA timezone string Returns: Dict with keys "data", "_meta". "data" contains name, altitude, azimuth. """ location, time_info = process_location_and_time(lon, lat, time, time_zone) result = await asyncio.to_thread(get_constellation_center, constellation_name, location, time_info) return format_response(result)
• src/celestial.py:211-258 (helper)

  Core helper function that calculates the altitude and azimuth of a constellation's center. Uses pre-loaded JSON data for centers, with fallbacks to specific stars or SIMBAD queries.
  def get_constellation_center( constellation_name: str, observer_location: EarthLocation, time: Union[Time, datetime] ) -> Dict[str, Any]: """ Return the apparent Alt/Az of a constellation's representative center using local data. """ # Convert local time to UTC if input is datetime if isinstance(time, datetime): if time.tzinfo is None: raise ValueError("Input datetime must be timezone-aware for local time.") time = Time(time.astimezone(pytz.UTC)) centers = _load_constellation_centers() centers_map = {item["name"].lower(): item for item in centers} key = constellation_name.lower() if key in centers_map: ra = float(centers_map[key]["ra"]) dec = float(centers_map[key]["dec"]) center_coord = SkyCoord(ra=ra*u.deg, dec=dec*u.deg, frame='icrs') else: fallback = { "ursa major": "Alioth", "ursa minor": "Polaris", "cassiopeia": "Schedar", "southern cross": "Acrux", "crux": "Acrux", "orion": "Betelgeuse", "scorpius": "Antares", "leo": "Regulus", "gemini": "Pollux", "taurus": "Aldebaran", "canis major": "Sirius" } if key in fallback: center_coord = _resolve_simbad_object(fallback[key]) else: center_coord = _resolve_simbad_object(constellation_name) altaz_frame = AltAz(obstime=time, location=observer_location) altaz = center_coord.transform_to(altaz_frame) return { "name": constellation_name, "altitude": float(altaz.alt.deg), "azimuth": float(altaz.az.deg) }
• src/celestial.py:281-291 (helper)

  Utility function to load precomputed constellation center coordinates from JSON file, with caching.
  def _load_constellation_centers(): global CONSTELLATIONS_CACHE if CONSTELLATIONS_CACHE is not None: return CONSTELLATIONS_CACHE data_path = os.path.join(os.path.dirname(__file__), 'data/constellation_centers.json') try: with open(data_path, 'r') as f: CONSTELLATIONS_CACHE = json.load(f) except FileNotFoundError: CONSTELLATIONS_CACHE = [] return CONSTELLATIONS_CACHE
• src/functions/celestial/impl.py:5-123 (registration)

  The @mcp.tool() decorator registers the get_constellation function as an MCP tool.
  from src.server_instance import mcp from src.celestial import celestial_pos, celestial_rise_set, calculate_moon_info, get_visible_planets, get_constellation_center, calculate_nightly_forecast from src.utils import process_location_and_time from src.response import format_response @mcp.tool() async def get_celestial_pos( celestial_object: str, lon: float, lat: float, time: str, time_zone: str ) -> Dict[str, Any]: """Calculate the altitude and azimuth angles of a celestial object. Args: celestial_object: Name of object (e.g. "sun", "moon", "andromeda") lon: Observer longitude in degrees lat: Observer latitude in degrees time: Observation time string "YYYY-MM-DD HH:MM:SS" time_zone: IANA timezone string Returns: Dict with keys "data", "_meta". "data" contains "altitude" and "azimuth" (degrees). """ location, time_info = process_location_and_time(lon, lat, time, time_zone) # Run synchronous celestial calculations in a separate thread to avoid blocking the event loop alt, az = await asyncio.to_thread(celestial_pos, celestial_object, location, time_info) return format_response({ "altitude": alt, "azimuth": az }) @mcp.tool() async def get_celestial_rise_set( celestial_object: str, lon: float, lat: float, time: str, time_zone: str ) -> Dict[str, Any]: """Calculate the rise and set times of a celestial object. Args: celestial_object: Name of object (e.g. "sun", "moon", "andromeda") lon: Observer longitude in degrees lat: Observer latitude in degrees time: Date string "YYYY-MM-DD HH:MM:SS" time_zone: IANA timezone string Returns: Dict with keys "data", "_meta". "data" contains "rise_time" and "set_time". """ location, time_info = process_location_and_time(lon, lat, time, time_zone) # Run synchronous celestial calculations in a separate thread rise_time, set_time = await asyncio.to_thread(celestial_rise_set, celestial_object, location, time_info) return format_response({ "rise_time": rise_time.isoformat() if rise_time else None, "set_time": set_time.isoformat() if set_time else None }) @mcp.tool() async def get_moon_info( time: str, time_zone: str ) -> Dict[str, Any]: """Get detailed information about the Moon's phase and position. Args: time: Date string "YYYY-MM-DD HH:MM:SS" time_zone: IANA timezone string Returns: Dict with keys "data", "_meta". "data" contains illumination, phase_name, age_days, etc. """ try: # Try standard format first dt = datetime.strptime(time, "%Y-%m-%d %H:%M:%S") except ValueError: try: # Try ISO format dt = datetime.fromisoformat(time) except ValueError: raise ValueError(f"Time string '{time}' matches neither '%Y-%m-%d %H:%M:%S' nor ISO format.") if dt.tzinfo is None: tz = pytz.timezone(time_zone) dt = tz.localize(dt) result = await asyncio.to_thread(calculate_moon_info, dt) return format_response(result) @mcp.tool() async def get_visible_planets( lon: float, lat: float, time: str, time_zone: str ) -> Dict[str, Any]: """Get a list of solar system planets currently visible (above horizon). Args: lon: Observer longitude in degrees lat: Observer latitude in degrees time: Observation time string "YYYY-MM-DD HH:MM:SS" time_zone: IANA timezone string Returns: Dict with keys "data", "_meta". "data" is a list of planet dicts (name, altitude, azimuth). """ location, time_info = process_location_and_time(lon, lat, time, time_zone) # Note: Function name collision with imported function 'get_visible_planets' # Use the imported function from src.celestial from src.celestial import get_visible_planets as calc_visible_planets planets = await asyncio.to_thread(calc_visible_planets, location, time_info) return format_response(planets) @mcp.tool()