SWLC MCP Server

Instructions

Implementation Reference

• src/swlc_mcp/server.py:1314-1327 (handler)

  MCP tool handler in call_tool() that invokes service.get_qlc_latest() and formats the response as text.

              elif name == "get_latest_qlc":
                  result = await lottery_service.get_qlc_latest()
                  if result:
                      return [types.TextContent(
                          type="text", 
                          text=f"""七乐彩最新开奖结果：
  期号：{result.period}
  开奖日期：{result.draw_date}
  基本号码：{' '.join(result.numbers)}
  特别号码：{' '.join(result.special_numbers or [])}
  奖池金额：{result.prize_pool or '暂无'}"""
                      )]
                  else:
                      return [types.TextContent(type="text", text="获取七乐彩数据失败")]

• src/swlc_mcp/server.py:1050-1058 (registration)

  Tool registration and schema definition for 'get_latest_qlc' in the list_tools() handler.

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

• src/swlc_mcp/server.py:293-357 (helper)

  SWLCService.get_qlc_latest(): Fetches latest QLC lottery result, prefers database cache, updates from web if stale, calls db.get_latest_qlc().

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

• src/swlc_mcp/database.py:311-338 (helper)

  Database method that queries the latest QLC result from the qilecai_results table in SQLite.

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