SWLC MCP Server

Instructions

Implementation Reference

• src/swlc_mcp/server.py:1185-1204 (handler)

  MCP tool handler implementation for 'sync_lottery_data'. Calls get_historical_data to sync data if needed and returns success message with synced periods count.
  elif name == "sync_lottery_data": lottery_type = arguments.get("lottery_type") periods = arguments.get("periods", 10) try: # 从网络获取数据并保存到数据库 results = await lottery_service.get_historical_data(lottery_type, periods) if results: # 记录同步日志 lottery_service.db.log_sync(lottery_type, len(results)) return [types.TextContent( type="text", text=f"成功同步{lottery_type}数据{len(results)}期到本地数据库" )] else: lottery_service.db.log_sync(lottery_type, 0, 'failed', '获取数据失败') return [types.TextContent(type="text", text=f"同步{lottery_type}数据失败")] except Exception as e: lottery_service.db.log_sync(lottery_type, 0, 'failed', str(e)) return [types.TextContent(type="text", text=f"同步{lottery_type}数据失败：{str(e)}")]
• src/swlc_mcp/server.py:1005-1025 (schema)

  Pydantic input schema for the 'sync_lottery_data' tool, defining parameters lottery_type (required enum) and optional periods (1-50, default 10).
  name="sync_lottery_data", description="同步指定彩票类型的最新数据到本地数据库", inputSchema={ "type": "object", "properties": { "lottery_type": { "type": "string", "enum": ["双色球", "福彩3D", "七乐彩", "快乐8"], "description": "彩票类型" }, "periods": { "type": "integer", "minimum": 1, "maximum": 50, "default": 10, "description": "同步期数" } }, "required": ["lottery_type"] } ),
• src/swlc_mcp/server.py:898-1057 (registration)

  Tool registration in the MCP server's list_tools() method, where 'sync_lottery_data' is included in the list of available tools.
  @server.list_tools() async def list_tools() -> List[types.Tool]: """列出所有可用工具""" return [ types.Tool( name="get_latest_ssq", description="获取双色球最新开奖结果", inputSchema={ "type": "object", "properties": {}, "required": [] } ), types.Tool( name="get_latest_3d", description="获取福彩3D最新开奖结果", inputSchema={ "type": "object", "properties": {}, "required": [] } ), types.Tool( name="get_latest_qlc", description="获取七乐彩最新开奖结果", inputSchema={ "type": "object", "properties": {}, "required": [] } ), types.Tool( name="get_latest_kl8", description="获取快乐8最新开奖结果", inputSchema={ "type": "object", "properties": {}, "required": [] } ), types.Tool( name="get_historical_data", description="获取指定彩票类型的历史开奖数据", inputSchema={ "type": "object", "properties": { "lottery_type": { "type": "string", "enum": ["双色球", "福彩3D", "七乐彩", "快乐8"], "description": "彩票类型" }, "periods": { "type": "integer", "minimum": 1, "maximum": 1000, "default": 10, "description": "获取期数" } }, "required": ["lottery_type"] } ), types.Tool( name="analyze_numbers", description="分析彩票号码统计信息，包括热号、冷号等", inputSchema={ "type": "object", "properties": { "lottery_type": { "type": "string", "enum": ["双色球", "福彩3D", "七乐彩", "快乐8"], "description": "彩票类型" }, "periods": { "type": "integer", "minimum": 5, "maximum": 1000, "default": 30, "description": "分析期数" } }, "required": ["lottery_type"] } ), types.Tool( name="generate_random_numbers", description="生成随机彩票号码推荐", inputSchema={ "type": "object", "properties": { "lottery_type": { "type": "string", "enum": ["双色球", "福彩3D", "七乐彩", "快乐8"], "description": "彩票类型" }, "count": { "type": "integer", "minimum": 1, "maximum": 10, "default": 1, "description": "生成组数" } }, "required": ["lottery_type"] } ), types.Tool( name="sync_lottery_data", description="同步指定彩票类型的最新数据到本地数据库", inputSchema={ "type": "object", "properties": { "lottery_type": { "type": "string", "enum": ["双色球", "福彩3D", "七乐彩", "快乐8"], "description": "彩票类型" }, "periods": { "type": "integer", "minimum": 1, "maximum": 50, "default": 10, "description": "同步期数" } }, "required": ["lottery_type"] } ), 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"] } ), types.Tool( name="get_database_info", description="获取本地数据库统计信息", inputSchema={ "type": "object", "properties": {}, "required": [] } ) ]
• src/swlc_mcp/server.py:412-587 (helper)

  SWLCService.get_historical_data: Core helper function that implements the data synchronization logic by fetching from network if local data is insufficient or stale, saving to database, and returning historical results. Called by the tool handler.
  async def get_historical_data(self, lottery_type: str, periods: int = 10) -> List[LotteryResult]: """获取历史开奖数据""" try: # 首先尝试从数据库获取 db_results = self.db.get_historical_data(lottery_type, periods) # 检查是否需要从网络更新数据 should_update = False if not db_results: # 数据库没有数据，需要从网络获取 should_update = True elif len(db_results) < periods: # 数据库数据不足，需要从网络获取 should_update = True else: # 检查最新数据的新鲜度 latest_result = db_results[0] # 最新的数据 if not self._is_data_fresh(latest_result.get('draw_date', ''), lottery_type): should_update = True if not should_update: logger.info(f"从本地数据库获取{lottery_type}历史数据") results = [] for item in db_results: if lottery_type == "双色球": result = LotteryResult( lottery_type="双色球", period=item['period'], draw_date=item['draw_date'], numbers=item['red_balls'], special_numbers=[item['blue_ball']], prize_pool=item['prize_pool'], sales_amount=item['sales_amount'] ) elif lottery_type == "福彩3D": result = LotteryResult( lottery_type="福彩3D", period=item['period'], draw_date=item['draw_date'], numbers=item['numbers'], sales_amount=item['sales_amount'] ) elif lottery_type == "七乐彩": result = LotteryResult( lottery_type="七乐彩", period=item['period'], draw_date=item['draw_date'], numbers=item['basic_numbers'], special_numbers=[item['special_number']], prize_pool=item['prize_pool'], sales_amount=item['sales_amount'] ) elif lottery_type == "快乐8": result = LotteryResult( lottery_type="快乐8", period=item['period'], draw_date=item['draw_date'], numbers=item['numbers'], prize_pool=item['prize_pool'], sales_amount=item['sales_amount'] ) else: continue results.append(result) return results # 从网络获取并保存数据 logger.info(f"从网络获取{lottery_type}历史数据") data = await self._fetch_lottery_data(lottery_type, periods) if not data or not data['result']: # 如果网络获取失败，尝试返回数据库中的可用数据 if db_results: logger.warning(f"网络获取{lottery_type}数据失败，返回数据库中的可用数据") return self._convert_db_results_to_lottery_results(db_results, lottery_type) return [] results = [] for item in data['result']: if lottery_type == "双色球": # 解析红球和蓝球 red_balls = item['red'].split(',') blue_ball = item['blue'] # 保存到数据库 self.db.save_ssq_result( period=item['code'], draw_date=item['date'], red_balls=red_balls, blue_ball=blue_ball ) result = LotteryResult( lottery_type="双色球", period=item['code'], draw_date=item['date'], numbers=red_balls, special_numbers=[blue_ball] ) elif lottery_type == "福彩3D": # 解析3D号码 numbers = item['red'].split(',') # 保存到数据库 self.db.save_3d_result( period=item['code'], draw_date=item['date'], numbers=numbers ) result = LotteryResult( lottery_type="福彩3D", period=item['code'], draw_date=item['date'], numbers=numbers ) elif lottery_type == "七乐彩": # 解析基本号码和特别号码 basic_numbers = item['red'].split(',') special_number = item['blue'] # 保存到数据库 self.db.save_qlc_result( period=item['code'], draw_date=item['date'], basic_numbers=basic_numbers, special_number=special_number ) result = LotteryResult( lottery_type="七乐彩", period=item['code'], draw_date=item['date'], numbers=basic_numbers, special_numbers=[special_number] ) elif lottery_type == "快乐8": # 解析快乐8号码 numbers = item['red'].split(',') # 保存到数据库 self.db.save_kl8_result( period=item['code'], draw_date=item['date'], numbers=numbers ) result = LotteryResult( lottery_type="快乐8", period=item['code'], draw_date=item['date'], numbers=numbers ) else: continue results.append(result) return results except Exception as e: logger.error(f"获取{lottery_type}历史数据失败: {e}") # 如果出错，尝试返回数据库中的可用数据 try: db_results = self.db.get_historical_data(lottery_type, periods) if db_results: logger.warning(f"返回数据库中的{lottery_type}数据作为备选") return self._convert_db_results_to_lottery_results(db_results, lottery_type) except: pass return []