product_search

Instructions

Implementation Reference

• src/biggo_mcp_server/tools/product_search.py:13-31 (handler)

  The main handler function for the 'product_search' tool. It accepts a 'query' string via Annotated Field, retrieves the server setting from the context, calls ProductSearchService.search(), and returns a slim JSON dump of the ProductSearchToolResponse.

  async def product_search(
      ctx: Context,
      query: Annotated[
          str,
          Field(
              description="""Search query""",
              examples=["iphone", "護唇膏"],
          ),
      ],
  ) -> str:
      """Product Search"""
      logger.info("product search, query: %s", query)
  
      setting = get_setting(ctx)
      service = ProductSearchService(setting)
      ret = await service.search(query)
  
      return ProductSearchToolResponse(product_search_result=ret).slim_dump()

• src/biggo_mcp_server/services/product_search.py:11-43 (helper)

  ProductSearchService contains the core business logic: calls the BigGo API (https://api.biggo.com/api/v1/spa/search/{query}/product) with domain/region headers, parses the response into ProductSearchAPIRet, generates real product URLs, and optionally shortens URLs.

  class ProductSearchService:
      def __init__(self, setting: BigGoMCPSetting):
          self._setting = setting
  
      async def search(self, query: str) -> ProductSearchAPIRet:
          url = f"https://api.biggo.com/api/v1/spa/search/{query}/product"
  
          headers = {
              "Content-Type": "application/json",
              "site": self._setting.domain.value,
              "region": self._setting.region.value.lower(),
          }
          logger.debug("product search, url: %s, headers: %s", url, headers)
  
          async with ClientSession() as session:
              async with session.get(url, headers=headers) as resp:
                  if resp.status >= 400:
                      err_msg = f"product search api error: {await resp.text()}"
                      logger.error(err_msg)
                      raise ValueError(err_msg)
  
                  data = ProductSearchAPIRet.model_validate(await resp.json())
  
          data.generate_r_link(self._setting.domain)
  
          if self._setting.short_url_endpoint is not None:
              all_urls = data.get_all_urls()
              url_map = await generate_short_url(
                  list(all_urls), self._setting.short_url_endpoint
              )
              data.replace_urls(url_map)
  
          return data

• src/biggo_mcp_server/lib/server_setup.py:14-31 (registration)

  Registration: The tool is registered via 'server.add_tool(product_search)' where product_search is imported from ..tools.product_search.

  async def create_server(setting: BigGoMCPSetting) -> BigGoMCPServer:
      server = BigGoMCPServer(setting)
      setup_logging(setting.log_level)
  
      # Product Search
      server.add_tool(product_search)
  
      # Price History
      # server.add_tool(price_history_graph)
      # server.add_tool(price_history_with_history_id)
      server.add_tool(price_history_with_url)
  
      # Spec Search
      # server.add_tool(spec_indexes)
      # server.add_tool(spec_mapping)
      # server.add_tool(spec_search)
  
      return server

• src/biggo_mcp_server/types/api_ret/product_search.py:82-131 (schema)

  Pydantic model ProductSearchAPIRet defines the API response schema (list of ListItem, low_price, high_price) with helper methods to generate real URLs, collect all URLs, and replace them with short URLs.

  class ProductSearchAPIRet(BaseModel):
      # result: bool
      # total: int
      # total_page: int
      # pure_total: int
      # ec_count: int
      # mall_count: int
      # bid_count: int
      # size: int
      # took: int
      # is_shop: bool
      # is_suggest_query: bool
      # is_ypa: bool
      # is_adsense: bool
      # q_suggest: str
      # arr_suggest: List[str]
      # offline_count: int
      # spam_count: int
      # promo: List
      # filter: Dict[str, Any]
      # top_ad_count: int
      # group: Any
      # recommend_group: List
      list: List[ListItem] = Field(default_factory=list)
      # biggo_c: List[BiggoCItem]
      low_price: float = 0
      high_price: float = 0
  
      def generate_r_link(self, domain: Domains):
          for product in self.list:
              product.url = f"https://{domain.value}{product.affurl}"
              product.affurl = None
  
      def get_all_urls(self) -> set[str]:
          product_urls = {product.url for product in self.list}
          image_urls = {product.image for product in self.list}
          return product_urls | image_urls
  
      def replace_urls(self, url_map: dict[str, str]):
          """Replace urls in the list with the new urls
  
          Args:
              url_map (dict[str, str]): The map of old urls to new urls
          """
          for product in self.list:
              if product.url in url_map:
                  product.url = url_map[product.url]
              if product.image in url_map:
                  product.image = url_map[product.image]

• src/biggo_mcp_server/types/responses.py:13-50 (schema)

  ProductSearchToolResponse wraps the ProductSearchAPIRet result and includes post-initialization logic that sets a 'reason' when no results are found, or 'display_rules' when results exist, and uses slim_dump() for JSON serialization.

  class BaseToolResponse(BaseModel):
      def slim_dump(self) -> str:
          return self.model_dump_json(exclude_none=True, exclude_defaults=True)
  
  
  class ProductSearchToolResponse(BaseToolResponse):
      product_search_result: ProductSearchAPIRet
      reason: str | None = None
      display_rules: str | None = None
  
      @model_validator(mode="after")
      def post_init(self) -> Self:
          if len(self.product_search_result.list) == 0:
              self.reason = """
  No results found. Possible reasons:
  1. This search is much more complex than a simple product search.
  2. The user is asking things related to product specifications.
  
  If the problems might be related to the points listed above,
  please use the 'spec_search' tool and try again.
              """
              return self
  
          # add display rules if result is not empty
          else:
              self.display_rules = """
  As a product researcher, you need to find the most relavent product and present them in utmost detail.
  Without following the rules listed bellow, the output will become useless, you must follow the rules before responding to the user.
  All rules must be followed strictly.
  
  Here are a list of rules you must follow:
  Rule 1: Product image must be included when available, url is located in each object inside 'specs.images' field.
  Rule 2: If no avaliable image exist, ignore the image field completely, don't even write anything image related for that single product.
  Rule 3: Product urls must be included so that the user can by the product with a simple click if possible.
  Rule 4: Display more then one relavent product if possible, having multiple choices is a good thing.
              """
  
          return self