SWLC MCP Server

Instructions

Implementation Reference

• src/swlc_mcp/server.py:179-241 (handler)

  SWLCService.get_ssq_latest(): Primary handler for getting latest SSQ results. Checks local DB first (via get_latest_ssq helper), fetches from network if stale, saves/updates DB, returns formatted LotteryResult.

  async def get_ssq_latest(self) -> Optional[LotteryResult]:
      """获取双色球最新开奖结果"""
      try:
          # 首先尝试从数据库获取
          db_result = self.db.get_latest_ssq()
          
          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]
                  
                  # 解析红球和蓝球
                  red_balls = result_data['red'].split(',')
                  blue_ball = result_data['blue']
                  
                  # 格式化奖池金额
                  pool_money = result_data.get('poolmoney', '')
                  if pool_money and pool_money.isdigit():
                      pool_money = f"{int(pool_money) / 100000000:.2f}亿元"
                  
                  # 格式化销售金额
                  sales = result_data.get('sales', '')
                  if sales and sales.isdigit():
                      sales = f"{int(sales) / 100000000:.2f}亿元"
                  
                  # 保存到数据库
                  self.db.save_ssq_result(
                      period=result_data['code'],
                      draw_date=result_data['date'],
                      red_balls=red_balls,
                      blue_ball=blue_ball,
                      prize_pool=pool_money,
                      sales_amount=sales
                  )
                  
                  # 更新号码统计
                  self.db.update_number_statistics('双色球', red_balls + [blue_ball])
                  
                  return LotteryResult(
                      lottery_type="双色球",
                      period=result_data['code'],
                      draw_date=result_data['date'],
                      numbers=red_balls,
                      special_numbers=[blue_ball],
                      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['red_balls'],
                  special_numbers=[db_result['blue_ball']],
                  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/server.py:1032-1040 (registration)

  MCP tool registration for 'get_latest_ssq' in server.list_tools(), including empty input schema.

  types.Tool(
      name="get_latest_ssq",
      description="获取双色球最新开奖结果",
      inputSchema={
          "type": "object",
          "properties": {},
          "required": []
      }
  ),

• src/swlc_mcp/database.py:255-282 (helper)

  LotteryDatabase.get_latest_ssq(): Helper method that queries the SQLite database for the most recent SSQ result (latest period).

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

• src/swlc_mcp/server.py:1285-1298 (handler)

  MCP server.call_tool() handler dispatch for 'get_latest_ssq': Calls SWLCService.get_ssq_latest() and formats response as TextContent.

              if name == "get_latest_ssq":
                  result = await lottery_service.get_ssq_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 '暂无'}
  销售金额：{result.sales_amount or '暂无'}"""
                      )]
                  else:
                      return [types.TextContent(type="text", text="获取双色球数据失败")]