search_ads

Instructions

Implementation Reference

• lbc_mcp_server.py:75-75 (registration)

  The @mcp.tool() decorator registers 'search_ads' as an MCP tool on the FastMCP 'leboncoin' server instance.

  @mcp.tool()

• lbc_mcp_server.py:76-144 (handler)

  The search_ads function (decorated with @mcp.tool()) implements the core logic: accepts optional filters (text, category, region, sort, limit, page, price_min, price_max), resolves enums from the lbc library, enforces rate limiting via _check_rate_limit(), calls client.search() with the constructed kwargs, and returns a JSON string with results and metadata.

  def search_ads(
      text: Annotated[str, "Mots-clés de recherche (ex: 'vélo', 'iPhone 14')"] = "",
      category: Annotated[
          str,
          "Catégorie (ex: 'LOISIRS_VELOS', 'VEHICULES_VOITURES', 'IMMOBILIER_VENTES_IMMOBILIERES'). "
          "Valeur vide = toutes catégories.",
      ] = "",
      region: Annotated[
          str,
          "Région (ex: 'ILE_DE_FRANCE', 'AUVERGNE_RHONE_ALPES', 'BRETAGNE'). Vide = France entière.",
      ] = "",
      sort: Annotated[
          Literal["NEWEST", "OLDEST", "CHEAPEST", "EXPENSIVE", "RELEVANCE"],
          "Tri des résultats.",
      ] = "NEWEST",
      limit: Annotated[int, "Nombre max d'annonces à retourner (1-35)."] = 10,
      page: Annotated[int, "Numéro de page (commence à 1)."] = 1,
      price_min: Annotated[int, "Prix minimum en euros (0 = pas de minimum)."] = 0,
      price_max: Annotated[int, "Prix maximum en euros (0 = pas de maximum)."] = 0,
  ) -> str:
      """Recherche des annonces Leboncoin avec filtres. Limité à 10 requêtes/heure (DataDome)."""
      if err := _check_rate_limit():
          return err
      client = _get_client()
  
      kwargs: dict = {
          "sort": Sort[sort],
          "limit": min(max(limit, 1), 35),
          "page": page,
      }
  
      if text:
          kwargs["text"] = text
  
      if category:
          try:
              kwargs["category"] = Category[category]
          except KeyError:
              valid = [e.name for e in Category][:20]
              return json.dumps({"error": f"Catégorie '{category}' inconnue. Exemples: {valid}"})
  
      if region:
          try:
              kwargs["locations"] = Region[region]
          except KeyError:
              valid = [e.name for e in Region][:10]
              return json.dumps({"error": f"Région '{region}' inconnue. Exemples: {valid}"})
  
      if price_min > 0 or price_max > 0:
          pmin = price_min if price_min > 0 else 0
          pmax = price_max if price_max > 0 else 0
          if pmax > 0:
              kwargs["price"] = (pmin, pmax)
          else:
              kwargs["price"] = (pmin, 9_999_999)
  
      try:
          results = client.search(**kwargs)
      except DatadomeError as e:
          return json.dumps({"error": f"DataDome: {e}. Utilisez un réseau résidentiel."})
  
      ads = [_ad_to_dict(ad) for ad in results.ads]
      return json.dumps({
          "total": results.total,
          "max_pages": results.max_pages,
          "page": page,
          "count": len(ads),
          "ads": ads,
      }, ensure_ascii=False)

• lbc_mcp_server.py:76-95 (schema)

  Input schema defined via type annotations on search_ads parameters: text (str), category (str), region (str), sort (Literal['NEWEST','OLDEST','CHEAPEST','EXPENSIVE','RELEVANCE']), limit (int, 1-35), page (int), price_min (int), price_max (int). All have Annotated descriptions used by FastMCP for schema generation.

  def search_ads(
      text: Annotated[str, "Mots-clés de recherche (ex: 'vélo', 'iPhone 14')"] = "",
      category: Annotated[
          str,
          "Catégorie (ex: 'LOISIRS_VELOS', 'VEHICULES_VOITURES', 'IMMOBILIER_VENTES_IMMOBILIERES'). "
          "Valeur vide = toutes catégories.",
      ] = "",
      region: Annotated[
          str,
          "Région (ex: 'ILE_DE_FRANCE', 'AUVERGNE_RHONE_ALPES', 'BRETAGNE'). Vide = France entière.",
      ] = "",
      sort: Annotated[
          Literal["NEWEST", "OLDEST", "CHEAPEST", "EXPENSIVE", "RELEVANCE"],
          "Tri des résultats.",
      ] = "NEWEST",
      limit: Annotated[int, "Nombre max d'annonces à retourner (1-35)."] = 10,
      page: Annotated[int, "Numéro de page (commence à 1)."] = 1,
      price_min: Annotated[int, "Prix minimum en euros (0 = pas de minimum)."] = 0,
      price_max: Annotated[int, "Prix maximum en euros (0 = pas de maximum)."] = 0,
  ) -> str:

• lbc_mcp_server.py:35-48 (helper)

  Helper function _check_rate_limit() enforces a sliding window rate limit (max 10 searches per hour) to avoid DataDome flags. Returns a JSON error string if limit exceeded, otherwise None.

  def _check_rate_limit() -> str | None:
      """Retourne un message d'erreur JSON si la limite est atteinte, sinon None."""
      now = time.monotonic()
      while _search_timestamps and now - _search_timestamps[0] > _SEARCH_WINDOW:
          _search_timestamps.popleft()
      if len(_search_timestamps) >= _SEARCH_LIMIT:
          oldest = _search_timestamps[0]
          wait = int(_SEARCH_WINDOW - (now - oldest)) + 1
          return json.dumps({
              "error": f"Rate limit atteint ({_SEARCH_LIMIT} recherches/heure). "
                       f"Réessayez dans {wait // 60} min {wait % 60} s."
          })
      _search_timestamps.append(now)
      return None

• lbc_mcp_server.py:58-72 (helper)

  Helper function _ad_to_dict() converts an lbc ad object into a JSON-serializable dictionary with id, subject, price, url, city, zipcode, department, region, category, first_publication_date, has_phone, and images_count.

  def _ad_to_dict(ad) -> dict:
      return {
          "id": ad.id,
          "subject": ad.subject,
          "price": ad.price,
          "url": ad.url,
          "city": ad.location.city,
          "zipcode": ad.location.zipcode,
          "department": ad.location.department_name,
          "region": ad.location.region_name,
          "category": ad.category_name,
          "first_publication_date": str(ad.first_publication_date) if ad.first_publication_date else None,
          "has_phone": ad.has_phone,
          "images_count": len(ad.images) if ad.images else 0,
      }