SWLC MCP Server

Instructions

Implementation Reference

• src/swlc_mcp/server.py:705-856 (handler)

  Core implementation of the force_sync_data tool handler in SWLCService. Fetches lottery results from official CWL API using _fetch_lottery_data and saves them to the local SQLite database via type-specific save methods, bypassing freshness checks. Handles formatting of prize pool and sales data, updates number statistics, and returns sync summary.
  async def force_sync_data(self, lottery_type: str, periods: int = 20) -> Dict[str, Any]: """ 强制同步指定彩票类型的数据 Args: lottery_type: 彩票类型 periods: 同步的期数 Returns: Dict: 同步结果信息 """ try: logger.info(f"开始强制同步{lottery_type}数据，期数: {periods}") # 从网络获取最新数据 data = await self._fetch_lottery_data(lottery_type, periods) if not data or not data['result']: return { "success": False, "message": f"网络获取{lottery_type}数据失败", "lottery_type": lottery_type, "periods": periods } # 统计同步的数据 synced_count = 0 for item in data['result']: try: if lottery_type == "双色球": red_balls = item['red'].split(',') blue_ball = item['blue'] # 格式化奖池金额 pool_money = item.get('poolmoney', '') if pool_money and pool_money.isdigit(): pool_money = f"{int(pool_money) / 100000000:.2f}亿元" # 格式化销售金额 sales = item.get('sales', '') if sales and sales.isdigit(): sales = f"{int(sales) / 100000000:.2f}亿元" # 保存到数据库 if self.db.save_ssq_result( period=item['code'], draw_date=item['date'], red_balls=red_balls, blue_ball=blue_ball, prize_pool=pool_money, sales_amount=sales ): synced_count += 1 # 更新号码统计 self.db.update_number_statistics('双色球', red_balls + [blue_ball]) elif lottery_type == "福彩3D": numbers = item['red'].split(',') # 格式化销售金额 sales = item.get('sales', '') if sales and sales.isdigit(): sales = f"{int(sales) / 10000:.1f}万元" # 保存到数据库 if self.db.save_3d_result( period=item['code'], draw_date=item['date'], numbers=numbers, sales_amount=sales ): synced_count += 1 # 更新号码统计 self.db.update_number_statistics('福彩3D', numbers) elif lottery_type == "七乐彩": basic_numbers = item['red'].split(',') special_number = item['blue'] # 格式化奖池金额 pool_money = item.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 = item.get('sales', '') if sales and sales.isdigit(): sales = f"{int(sales) / 10000:.1f}万元" # 保存到数据库 if self.db.save_qlc_result( period=item['code'], draw_date=item['date'], basic_numbers=basic_numbers, special_number=special_number, prize_pool=pool_money, sales_amount=sales ): synced_count += 1 # 更新号码统计 self.db.update_number_statistics('七乐彩', basic_numbers + [special_number]) elif lottery_type == "快乐8": numbers = item['red'].split(',') # 格式化奖池金额 pool_money = item.get('poolmoney', '') if pool_money and pool_money.replace('.', '').isdigit(): pool_money = f"{float(pool_money) / 10000:.2f}万元" # 格式化销售金额 sales = item.get('sales', '') if sales and sales.isdigit(): sales = f"{int(sales) / 10000:.1f}万元" # 保存到数据库 if self.db.save_kl8_result( period=item['code'], draw_date=item['date'], numbers=numbers, prize_pool=pool_money, sales_amount=sales ): synced_count += 1 # 更新号码统计 self.db.update_number_statistics('快乐8', numbers) except Exception as e: logger.warning(f"保存{item['code']}期数据失败: {e}") continue logger.info(f"{lottery_type}数据同步完成，成功同步{synced_count}期") return { "success": True, "message": f"{lottery_type}数据同步成功", "lottery_type": lottery_type, "periods": periods, "synced_count": synced_count, "total_available": len(data['result']) } except Exception as e: logger.error(f"强制同步{lottery_type}数据失败: {e}") return { "success": False, "message": f"同步失败: {str(e)}", "lottery_type": lottery_type, "periods": periods }
• src/swlc_mcp/server.py:1185-1206 (registration)

  MCP tool registration in @server.list_tools(). Defines the tool name, description, and input schema (lottery_type required enum, periods optional 1-1000 default 20).
  types.Tool( name="force_sync_data", description="强制同步指定彩票类型的最新数据到本地数据库", inputSchema={ "type": "object", "properties": { "lottery_type": { "type": "string", "enum": ["双色球", "福彩3D", "七乐彩", "快乐8"], "description": "彩票类型" }, "periods": { "type": "integer", "minimum": 1, "maximum": 1000, "default": 20, "description": "同步期数" } }, "required": ["lottery_type"] } ),
• src/swlc_mcp/server.py:1450-1468 (handler)

  MCP server @server.call_tool() dispatcher branch for force_sync_data. Parses arguments, calls SWLCService.force_sync_data, formats success/error TextContent response.
  elif name == "force_sync_data": lottery_type = arguments.get("lottery_type") periods = arguments.get("periods", 20) try: sync_result = await lottery_service.force_sync_data(lottery_type, periods) if sync_result["success"]: return [types.TextContent( type="text", text=f"成功强制同步{sync_result['lottery_type']}数据{sync_result['synced_count']}期到本地数据库" )] else: return [types.TextContent( type="text", text=f"强制同步{sync_result['lottery_type']}数据失败: {sync_result['message']}" )] except Exception as e: return [types.TextContent(type="text", text=f"强制同步{lottery_type}数据失败：{str(e)}")]
• src/swlc_mcp/server.py:82-107 (helper)

  Helper method _fetch_lottery_data used by force_sync_data to query the official China Welfare Lottery API (https://www.cwl.gov.cn/.../findDrawNotice) with lottery code mapping and pagination.
  async def _fetch_lottery_data(self, lottery_type: str, page_size: int = 1) -> Optional[dict]: """通用的彩票数据获取方法""" try: lottery_code = self.lottery_codes.get(lottery_type) if not lottery_code: logger.error(f"不支持的彩票类型: {lottery_type}") return None params = { 'name': lottery_code, 'pageNo': 1, 'pageSize': page_size, 'systemType': 'PC' } response = await self.client.get(self.base_url, params=params) if response.status_code == 200: data = response.json() if data.get('state') == 0 and data.get('result'): return data return None except Exception as e: logger.error(f"获取{lottery_type}数据失败: {e}") return None
• src/swlc_mcp/api_server.py:381-419 (handler)

  HTTP API endpoint /api/force-sync/{lottery_type} that wraps the force_sync_data call for external HTTP access via FastAPI.
  @app.post("/api/force-sync/{lottery_type}") async def force_sync_lottery_data( lottery_type: str, periods: int = Query(20, ge=1, le=1000, description="强制同步期数") ): """强制同步彩票数据（忽略数据新鲜度检查）""" try: lottery_type_map = { "ssq": "双色球", "3d": "福彩3D", "qlc": "七乐彩", "kl8": "快乐8" } chinese_type = lottery_type_map.get(lottery_type) if not chinese_type: raise HTTPException(status_code=400, detail="不支持的彩票类型") # 强制同步数据 sync_result = await lottery_service.force_sync_data(chinese_type, periods) if sync_result["success"]: return { "success": True, "message": sync_result["message"], "data": { "lottery_type": chinese_type, "synced_count": sync_result["synced_count"], "total_available": sync_result["total_available"], "requested_periods": periods }, "timestamp": datetime.now().isoformat() } else: raise HTTPException(status_code=500, detail=sync_result["message"]) except Exception as e: logger.error(f"强制同步彩票数据失败: {e}") raise HTTPException(status_code=500, detail=str(e))