price_history_with_url

Instructions

Implementation Reference

• src/biggo_mcp_server/tools/price_history.py:67-84 (handler)

  The main handler function for the 'price_history_with_url' tool. It takes a product URL, calls the PriceHistoryService.history_with_url() method, and returns a PriceHistoryToolResponse with description and graph link.

  async def price_history_with_url(
      ctx: Context,
      url: Annotated[str, Field(description="Product URL")],
  ) -> str:
      """Product Price History With URL"""
  
      logger.info("price history with url, url: %s", url)
  
      setting = get_setting(ctx)
      service = PriceHistoryService(setting)
      ret = await service.history_with_url(url=url, days="180")
      if ret is None:
          return "No price history found"
  
      return PriceHistoryToolResponse(
          price_history_description=ret.description,
          price_history_graph=f"![Price History Graph]({ret.graph_link})",
      ).slim_dump()

• src/biggo_mcp_server/tools/price_history.py:68-71 (schema)

  Input schema for the tool: requires a 'url' parameter (string) annotated with Pydantic Field description.

      ctx: Context,
      url: Annotated[str, Field(description="Product URL")],
  ) -> str:
      """Product Price History With URL"""

• src/biggo_mcp_server/types/responses.py:60-72 (schema)

  Response schema (PriceHistoryToolResponse) used by price_history_with_url to format output with price_history_description and price_history_graph.

  class PriceHistoryToolResponse(BaseToolResponse):
      price_history_description: PriceHistoryAPIRet
      price_history_graph: str
      display_rules: str = """
  Field explanation:
  'price_history_description': includes detailed price history info.
  'price_history_graph': includes a markdown image link used for visualizing price history.
  
  Here are a list of rules you must follow:
  Rule 1: Both 'price_history_description' and the link provided at 'price_history_graph' field must be included in the output.
  Rule 2: Product url must be included, it can be found in 'price_history_description'
      """

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

  The tool is registered with the MCP server via server.add_tool(price_history_with_url) in create_server().

  server.add_tool(price_history_with_url)

• src/biggo_mcp_server/services/price_history.py:81-124 (helper)

  The core service method history_with_url() that expands the URL, resolves nindex and product ID, fetches price history from the BigGo API, optionally tries an alternative nindex for Shopee products, and returns a PriceHistoryRet with description and graph link.

  async def history_with_url(self, url: str, days: DAYS) -> PriceHistoryRet | None:
      real_url = await expand_url(url)
  
      if (nindex := await get_nindex_from_url(real_url)) is None:
          logger.warning("nindex not found, url: %s", real_url)
          return
  
      if (pid := await get_pid_from_url(nindex=nindex, url=real_url)) is None:
          logger.warning(
              "product id not found, nindex: %s, url: %s", nindex, real_url
          )
          return
  
      history_id = self._get_history_id(nindex=nindex, pid=pid)
      resp = await self._get_price_history(history_id=history_id, days=int(days))
  
      if resp is None and nindex in ["tw_mall_shopeemall", "tw_bid_shopee"]:
          nindex = (
              "tw_mall_shopeemall"
              if nindex == "tw_bid_shopee"
              else "tw_mall_shopeemall"
          )
          history_id = self._get_history_id(nindex=nindex, pid=pid)
          resp = await self._get_price_history(
              history_id=history_id,
              days=int(days),
          )
  
      if resp is not None:
          graph_link = self.graph_link(history_id)
  
          # replace urls with short urls
          if self._setting.short_url_endpoint is not None:
              all_urls = resp.get_all_urls()
              all_urls.add(graph_link)
              url_map = await generate_short_url(
                  list(all_urls), self._setting.short_url_endpoint
              )
              resp.replace_urls(url_map)
              graph_link = url_map.get(graph_link, graph_link)
  
          return PriceHistoryRet(description=resp, graph_link=graph_link)
      else:
          return None