mcp-server-public-transport

Instructions

Implementation Reference

• tools/no.py:181-214 (handler)

  The async handler function that executes the no_stop_departures tool logic, querying the Entur GraphQL API for upcoming departures from a stop place.
  async def no_stop_departures(stop_place_id: str, limit: int | None = 10) -> dict[str, object]: """ Args: stop_place_id: NSR StopPlace ID string. limit: Number of departures to fetch. Default: 10. Returns: GraphQL `data` with stopPlace + estimatedCalls. """ if not stop_place_id or not stop_place_id.strip(): raise ValueError("Parameter 'stop_place_id' must not be empty.") query = """ query StopDepartures($id: String!, $limit: Int!) { stopPlace(id: $id) { id name estimatedCalls(numberOfDepartures: $limit) { realtime aimedDepartureTime expectedDepartureTime destinationDisplay { frontText } quay { id name } serviceJourney { id line { id name publicCode transportMode } } } } } """ variables = {"id": stop_place_id.strip(), "limit": int(limit or 10)} logger.info("Entur stop departures: %s (limit=%s)", variables["id"], variables["limit"]) return await _post_graphql(query, variables)
• tools/no.py:177-180 (registration)

  The @mcp.tool decorator that registers the no_stop_departures tool with the MCP server.
  @mcp.tool( name="no_stop_departures", description="Upcoming departures for a StopPlace ID (e.g., 'NSR:StopPlace:58368')." )
• tools/no.py:181-188 (schema)

  Function signature and docstring defining input parameters and output type for the tool.
  async def no_stop_departures(stop_place_id: str, limit: int | None = 10) -> dict[str, object]: """ Args: stop_place_id: NSR StopPlace ID string. limit: Number of departures to fetch. Default: 10. Returns: GraphQL `data` with stopPlace + estimatedCalls. """
• tools/no.py:82-119 (helper)

  Helper function used by no_stop_departures to POST GraphQL queries to Entur API with retries.
  async def _post_graphql( query: str, variables: dict[str, object] | None = None, timeout: int = DEFAULT_TOTAL_TIMEOUT, tries: int = 3, ) -> dict[str, object]: """POST a GraphQL query to Entur Journey Planner v3 and return the `data` field.""" payload = {"query": query, "variables": variables or {}} for attempt in range(1, tries + 1): try: async with aiohttp.ClientSession(headers=COMMON_HEADERS, timeout=_make_timeout(timeout)) as session: async with session.post(NO_JP_BASE_URL, json=payload, timeout=_make_timeout(timeout)) as resp: # Retry on rate limit or server errors if resp.status == 429 or resp.status >= 500: text = await resp.text() if attempt < tries: await asyncio.sleep(0.5 * (2 ** (attempt - 1))) continue raise TransportAPIError(f"Entur GraphQL HTTP {resp.status}: {text}") if resp.status >= 400: text = await resp.text() raise TransportAPIError(f"Entur GraphQL HTTP {resp.status}: {text}") data = await resp.json() if "errors" in data and data["errors"]: raise TransportAPIError(f"Entur GraphQL errors: {data['errors']}") return data.get("data", {}) except (asyncio.TimeoutError, aiohttp.ServerTimeoutError) as e: if attempt < tries: await asyncio.sleep(0.5 * (2 ** (attempt - 1))) continue raise TransportAPIError(f"Entur GraphQL timeout after {tries} attempt(s): {e}") from e # Should never reach here raise TransportAPIError("Entur GraphQL: exhausted retries without response")
• tools/no.py:316-316 (registration)

  List of registered Norway tools returned by register_no_tools, including no_stop_departures.
  return ["no_search_places", "no_stop_departures", "no_trip", "no_nearest_stops"]