Search Hotels With Details

Instructions

Implementation Reference

• stays/mcp/server.py:312-319 (handler)

  MCP tool registration of search_hotels_with_details, wrapping _search_hotels_with_details_impl which constructs SearchHotelsWithDetailsParams and delegates to _execute_search_hotels_with_details_from_params.

  search_hotels_with_details = mcp.tool(
      name="search_hotels_with_details",
      annotations={
          "title": "Search Hotels With Details",
          "readOnlyHint": True,
          "idempotentHint": True,
      },
  )(_search_hotels_with_details_impl)

• stays/mcp/server.py:244-294 (handler)

  Core implementation function for the search_hotels_with_details tool. Builds params from individual arguments and delegates to the executor.

  def _search_hotels_with_details_impl(
      query: Annotated[str, Field()],
      check_in: Annotated[str, Field()],
      check_out: Annotated[str, Field()],
      max_hotels: Annotated[
          int,
          Field(
              ge=1,
              le=HARD_MAX_HOTELS_WITH_DETAILS,
              description=(f"Top-N hotels to enrich with detail. Hard cap = {HARD_MAX_HOTELS_WITH_DETAILS}. Default 5."),
          ),
      ] = CONFIG.default_max_hotels_with_details,
      adults: Annotated[int, Field(ge=1)] = CONFIG.default_adults,
      children: Annotated[int, Field(ge=0, le=8)] = CONFIG.default_children,
      child_ages: Annotated[list[int] | None, Field()] = None,
      currency: Annotated[str, Field(min_length=3, max_length=3)] = CONFIG.default_currency,
      sort_by: Annotated[SortByLiteral, Field()] = CONFIG.default_sort_by,
      hotel_class: Annotated[list[int] | None, Field()] = None,
      amenities: Annotated[list[str] | None, Field()] = None,
      brands: Annotated[list[str] | None, Field()] = None,
      min_guest_rating: Annotated[float | None, Field(ge=3.5, le=4.5)] = None,
      free_cancellation: bool = False,
      eco_certified: bool = False,
      special_offers: bool = False,
      price_min: Annotated[int | None, Field(ge=0)] = None,
      price_max: Annotated[int | None, Field(ge=0)] = None,
      property_type: Annotated[PropertyTypeLiteral, Field()] = "HOTELS",
  ) -> dict[str, Any]:
      """placeholder — overwritten via __doc__ assignment below."""
      params = SearchHotelsWithDetailsParams(
          query=query,
          check_in=check_in,
          check_out=check_out,
          max_hotels=max_hotels,
          adults=adults,
          children=children,
          child_ages=child_ages,
          currency=currency,
          sort_by=sort_by,
          hotel_class=hotel_class,
          amenities=amenities,
          brands=brands,
          min_guest_rating=min_guest_rating,
          free_cancellation=free_cancellation,
          eco_certified=eco_certified,
          special_offers=special_offers,
          price_min=price_min,
          price_max=price_max,
          property_type=property_type,
      )
      return _execute_search_hotels_with_details_from_params(params)

• stays/mcp/_executors.py:227-248 (handler)

  Executor that converts SearchHotelsWithDetailsParams to filters, calls search_with_details on the SearchHotels class, and serializes the enriched results.

  def _execute_search_hotels_with_details_from_params(
      params: SearchHotelsWithDetailsParams,
  ) -> dict[str, Any]:
      try:
          shp = SearchHotelsParams(**params.model_dump(exclude={"max_hotels"}))
          filters = _build_filters_from_search_params(shp)
          enriched = _get_search_hotels_cls()().search_with_details(filters, max_hotels=params.max_hotels)
          items = []
          for er in enriched:
              items.append(
                  {
                      "ok": er.ok,
                      "result": _serialize_hotel_result(er.result),
                      "detail": _serialize_hotel_detail(er.detail) if er.detail else None,
                      "error": er.error,
                      "error_kind": er.error_kind,
                      "is_retryable": er.is_retryable,
                  }
              )
          return {"success": True, "count": len(items), "items": items}
      except (BatchExecuteError, TransientBatchExecuteError) as e:
          return {"success": False, "error": f"{type(e).__name__}: {e}", "items": []}

• stays/mcp/_params.py:75-104 (schema)

  Pydantic model defining input parameters and validation for search_hotels_with_details, including the max_hotels field capped at HARD_MAX_HOTELS_WITH_DETAILS=15.

  class SearchHotelsWithDetailsParams(BaseModel):
      query: str = Field(description="City or property query.")
      check_in: str = Field(description="YYYY-MM-DD — REQUIRED.")
      check_out: str = Field(description="YYYY-MM-DD after check_in — REQUIRED.")
      max_hotels: int = Field(
          default=CONFIG.default_max_hotels_with_details,
          ge=1,
          le=HARD_MAX_HOTELS_WITH_DETAILS,
          description=f"Top-N to enrich. HARD CAP = {HARD_MAX_HOTELS_WITH_DETAILS}.",
      )
      adults: int = Field(default=CONFIG.default_adults, ge=1)
      children: int = Field(default=CONFIG.default_children, ge=0, le=8)
      child_ages: list[int] | None = Field(default=None)
      currency: str = Field(default=CONFIG.default_currency, min_length=3, max_length=3)
      sort_by: SortByLiteral = CONFIG.default_sort_by
      hotel_class: list[int] | None = None
      amenities: list[str] | None = None
      brands: list[str] | None = None
      min_guest_rating: float | None = Field(default=None, ge=3.5, le=4.5)
      free_cancellation: bool = False
      eco_certified: bool = False
      special_offers: bool = False
      price_min: int | None = Field(default=None, ge=0)
      price_max: int | None = Field(default=None, ge=0)
      property_type: PropertyTypeLiteral = "HOTELS"
  
      @model_validator(mode="after")
      def _child_ages_matches_children(self):
          _validate_child_ages(self.children, self.child_ages)
          return self

• stays/search/hotels.py:172-237 (helper)

  Core search-with-details logic on the SearchHotels class: runs a search, then fetches parallel details for top N results using a ThreadPoolExecutor.

  def search_with_details(self, filters: HotelSearchFilters, max_hotels: int = 5) -> list[EnrichedResult]:
      """Run ``search()``, then fetch detail for the first
      ``max_hotels`` results in parallel. Partial failures are reported
      per-hotel via ``EnrichedResult.error``; the batch never aborts
      on a single transient."""
      if filters.dates is None:
          raise ValueError(
              "search_with_details requires filters.dates so that detail "
              "responses can carry rate plans. Set dates on your HotelSearchFilters."
          )
      results = self.search(filters)
      top = results[:max_hotels]
      workers = min(self._detail_concurrency, max(1, len(top)))
      logger.info("enrich count=%d concurrency=%d", len(top), workers)
  
      def enrich_one(r: HotelResult) -> EnrichedResult:
          if not r.entity_key:
              logger.warning(
                  "enrich error hotel=%s kind=%s msg=%s",
                  r.name,
                  "fatal",
                  "missing entity_key",
              )
              return EnrichedResult(
                  result=r,
                  error="missing entity_key",
                  error_kind="fatal",
              )
          try:
              detail = self.get_details(
                  entity_key=r.entity_key,
                  dates=filters.dates,
                  location=filters.location,
                  currency=filters.currency,
              )
              return EnrichedResult(result=r, detail=detail)
          except TransientBatchExecuteError as e:
              logger.warning(
                  "enrich error hotel=%s kind=%s msg=%s",
                  r.name,
                  "transient",
                  f"{type(e).__name__}: {e}",
              )
              return EnrichedResult(
                  result=r,
                  error=f"{type(e).__name__}: {e}",
                  error_kind="transient",
              )
          except (BatchExecuteError, MissingHotelIdError) as e:
              logger.warning(
                  "enrich error hotel=%s kind=%s msg=%s",
                  r.name,
                  "fatal",
                  f"{type(e).__name__}: {e}",
              )
              return EnrichedResult(
                  result=r,
                  error=f"{type(e).__name__}: {e}",
                  error_kind="fatal",
              )
          # Unknown exceptions intentionally NOT caught — they propagate
          # so parser bugs / programmer errors surface instead of being
          # silently stringified into per-hotel error fields.
  
      with ThreadPoolExecutor(max_workers=workers) as ex:
          return list(ex.map(enrich_one, top))