BigGo MCP Server

Instructions

Implementation Reference

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

  The MCP tool handler function 'product_search' that executes the tool logic by calling ProductSearchService and returning a formatted JSON response.

  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/lib/server_setup.py:18-20 (registration)

  Registration of the 'product_search' tool handler on the BigGoMCPServer instance.

  # Product Search
  server.add_tool(product_search)

• src/biggo_mcp_server/types/responses.py:18-51 (schema)

  Output schema for the product_search tool: ProductSearchToolResponse Pydantic model wrapping the API result with additional fields.

  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

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

  Core data schema: ProductSearchAPIRet Pydantic model defining the structure of the BigGo product search API response.

  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

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

  Helper service class ProductSearchService that handles the HTTP request to BigGo API, validates response, generates links, 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