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"]