SWLC MCP Server

Instructions

Implementation Reference

• src/swlc_mcp/server.py:173-235 (handler)

  Primary handler function in SWLCService that executes the logic for getting the latest Shuangseqiu (SSQ) lottery results. Checks local DB first, fetches from network if stale, parses, saves, and returns formatted result.
  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:902-910 (registration)

  MCP tool registration in list_tools(), defining the tool name 'get_latest_ssq', description, and empty input schema.
  types.Tool( name="get_latest_ssq", description="获取双色球最新开奖结果", inputSchema={ "type": "object", "properties": {}, "required": [] } ),
• src/swlc_mcp/server.py:1063-1077 (handler)

  Dispatch handler in @server.call_tool() that handles the 'get_latest_ssq' tool call, invokes service.get_ssq_latest(), and formats the TextContent response.
  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="获取双色球数据失败")]
• src/swlc_mcp/database.py:255-282 (helper)

  Database helper method LotteryDatabase.get_latest_ssq() that queries the local SQLite DB for the most recent SSQ result.
  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:905-909 (schema)

  Input schema for get_latest_ssq tool: empty object (no parameters required).
  inputSchema={ "type": "object", "properties": {}, "required": [] }