Weibo MCP Server

Instructions

Get the current hot search topics on Weibo.
    
Returns:
    list[dict]: List of dictionaries containing hot search items

Implementation Reference

• src/mcp_server_weibo/server.py:70-81 (handler)

  MCP tool handler for 'get_trendings' that delegates to WeiboCrawler instance.

  @mcp.tool()
  async def get_trendings(
      ctx: Context, 
      limit: Annotated[int, Field(description="Maximum number of hot search items to return, defaults to 15", default=15)] = 15
      ) -> list[dict]:
      """
      Get the current hot search topics on Weibo.
          
      Returns:
          list[dict]: List of dictionaries containing hot search items
      """
      return await crawler.get_trendings(limit)

• src/mcp_server_weibo/schemas.py:75-89 (schema)

  Pydantic model defining the structure of TrendingItem returned by the tool.

  class TrendingItem(BaseModel):
      """
      Data model for a single hot search item on Weibo.
      
      Attributes:
          id (int): Rank of the search item
          trending (int): Popularity value of the hot search item
          description (str): The description of hot search item
          url (str): URL to the hot search item
      """
      id: int = Field()
      trending: int = Field()
      description: str = Field()
      url: str

• src/mcp_server_weibo/weibo.py:128-162 (helper)

  Core implementation of fetching Weibo trending topics in WeiboCrawler class, including API call and data parsing.

  async def get_trendings(self, limit: int = 15) -> list[TrendingItem]:
      """
      Get a list of hot search items from Weibo.
  
      Args:
          limit (int): Maximum number of hot search items to return, defaults to 15
  
      Returns:
          list[HotSearchItem]: List of HotSearchItem objects containing hot search information
      """
      try:
          params = {
              'containerid': f'106003type=25&t=3&disable_hot=1&filter_type=realtimehot',
          }
          encoded_params = urlencode(params)
          
          async with httpx.AsyncClient() as client:
              response = await client.get(f'{SEARCH_URL}?{encoded_params}', headers=DEFAULT_HEADERS)
              data = response.json()
              cards = data.get('data', {}).get('cards', [])
              if not cards:
                  return []
  
              hot_search_card = next((card for card in cards if 'card_group' in card and isinstance(card['card_group'], list)), None)
              if not hot_search_card or 'card_group' not in hot_search_card:
                  return []
  
              items = [item for item in hot_search_card['card_group'] if item.get('desc')]
              trending_items = list(map(lambda pair: self._to_trending_item({**pair[1], 'id': pair[0]}), enumerate(items[:limit])))
              return trending_items
      except httpx.HTTPError:
          self.logger.error(
              'Unable to fetch Weibo hot search list', exc_info=True)
          return []

• src/mcp_server_weibo/weibo.py:395-412 (helper)

  Helper method to transform raw API data into TrendingItem objects.

  def _to_trending_item(self, item: dict) -> TrendingItem:
      """
      Convert raw hot search item data to HotSearchItem object.
  
      Args:
          item (dict): Raw hot search item data from Weibo API
  
      Returns:
          HotSearchItem: Formatted hot search item information
      """
      extr_values = re.findall(r'\d+', str(item.get('desc_extr')))
      trending = int(extr_values[0]) if extr_values else 0
      return TrendingItem(
          id=item['id'],
          trending=trending,
          description=item['desc'],
          url=item.get('scheme', '')
      )