BigGo MCP Server

Instructions

Implementation Reference

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

  The main MCP tool handler function for 'product_search'. It receives the search query, instantiates the service with settings, performs the search, and returns a slim JSON dump of the response model.
  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:19-19 (registration)

  Registers the 'product_search' tool function with the BigGo MCP server during server setup.
  server.add_tool(product_search)
• src/biggo_mcp_server/types/responses.py:18-51 (schema)

  Pydantic schema for the tool's output response, wrapping the API results with optional reason and display rules, including a validator for empty results.
  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/services/product_search.py:15-43 (helper)

  Core helper method in ProductSearchService that performs HTTP GET to BigGo API, validates response with Pydantic, generates affiliate links, optionally shortens URLs, and returns the processed data.
  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/types/api_ret/product_search.py:82-131 (schema)

  Pydantic model for the raw API response from BigGo product search, including list of ListItem products, price ranges, and utility methods for link generation and URL replacement.
  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]