SWLC MCP Server

Instructions

Implementation Reference

• src/swlc_mcp/server.py:1059-1067 (registration)

  Registration of the MCP tool 'get_latest_kl8' in the list_tools() function, including name, description, and empty input schema.

  types.Tool(
      name="get_latest_kl8",
      description="获取快乐8最新开奖结果",
      inputSchema={
          "type": "object",
          "properties": {},
          "required": []
      }
  ),

• src/swlc_mcp/server.py:1329-1342 (handler)

  MCP tool handler in call_tool() for 'get_latest_kl8': calls SWLCService.get_kl8_latest() and formats the text response.

              elif name == "get_latest_kl8":
                  result = await lottery_service.get_kl8_latest()
                  if result:
                      return [types.TextContent(
                          type="text",
                          text=f"""快乐8最新开奖结果：
  期号：{result.period}
  开奖日期：{result.draw_date}
  开奖号码：{' '.join(result.numbers)}
  奖池金额：{result.prize_pool or '暂无'}
  销售金额：{result.sales_amount or '暂无'}"""
                      )]
                  else:
                      return [types.TextContent(type="text", text="获取快乐8数据失败")]

• src/swlc_mcp/server.py:359-416 (handler)

  Main tool logic in SWLCService.get_kl8_latest(): checks DB for latest KL8 result, fetches from network if outdated, saves, updates stats, returns formatted result.

  async def get_kl8_latest(self) -> Optional[LotteryResult]:
      """获取快乐8最新开奖结果"""
      try:
          # 首先尝试从数据库获取
          db_result = self.db.get_latest_kl8()
          
          if self._should_update_from_network(db_result, "快乐8"):
              logger.info("从网络获取快乐8数据")
              data = await self._fetch_lottery_data('快乐8')
              if data and data['result']:
                  result_data = data['result'][0]
                  
                  # 解析快乐8号码 (20个号码)
                  numbers = result_data['red'].split(',')
                  
                  # 格式化奖池金额
                  pool_money = result_data.get('poolmoney', '')
                  if pool_money and pool_money.replace('.', '').isdigit():
                      pool_money = f"{float(pool_money) / 10000:.2f}万元"
                  
                  # 格式化销售金额
                  sales = result_data.get('sales', '')
                  if sales and sales.isdigit():
                      sales = f"{int(sales) / 10000:.1f}万元"
                  
                  # 保存到数据库
                  self.db.save_kl8_result(
                      period=result_data['code'],
                      draw_date=result_data['date'],
                      numbers=numbers,
                      prize_pool=pool_money,
                      sales_amount=sales
                  )
                  
                  # 更新号码统计
                  self.db.update_number_statistics('快乐8', numbers)
                  
                  return LotteryResult(
                      lottery_type="快乐8",
                      period=result_data['code'],
                      draw_date=result_data['date'],
                      numbers=numbers,
                      prize_pool=pool_money,
                      sales_amount=sales
                  )
          else:
              logger.info("从本地数据库获取快乐8数据")
              return LotteryResult(
                  lottery_type="快乐8",
                  period=db_result['period'],
                  draw_date=db_result['draw_date'],
                  numbers=db_result['numbers'],
                  prize_pool=db_result['prize_pool'],
                  sales_amount=db_result['sales_amount']
              )
      except Exception as e:
          logger.error(f"获取快乐8数据失败: {e}")
          return None

• src/swlc_mcp/database.py:340-366 (helper)

  Database helper method LotteryDatabase.get_latest_kl8(): queries the kuaile8_results table for the latest KL8 result and returns formatted dict.

  def get_latest_kl8(self) -> Optional[Dict[str, Any]]:
      """获取最新快乐8开奖结果"""
      try:
          with sqlite3.connect(self.db_path) as conn:
              cursor = conn.cursor()
              
              cursor.execute("""
                  SELECT period, draw_date, numbers, prize_pool, sales_amount
                  FROM kuaile8_results 
                  ORDER BY period DESC 
                  LIMIT 1
              """)
              
              result = cursor.fetchone()
              if result:
                  return {
                      'period': result[0],
                      'draw_date': result[1],
                      'numbers': json.loads(result[2]),
                      'prize_pool': result[3],
                      'sales_amount': result[4]
                  }
              return None
              
      except Exception as e:
          logger.error(f"获取最新快乐8数据失败: {e}")
          return None

• src/swlc_mcp/server.py:1062-1066 (schema)

  Input schema for 'get_latest_kl8' tool: empty object (no parameters required).

  inputSchema={
      "type": "object",
      "properties": {},
      "required": []
  }