get_canteen_data
Retrieve cafeteria dining attendance data, including breakfast and lunch counts, within a specified date range using the MCP Canteen Server.
Instructions
获取餐厅就餐人数数据
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| end_date | No | ||
| period | No | ||
| start_date | No |
Implementation Reference
- src/mcp_server_canteen/server.py:176-240 (handler)Main handler function implementing the 'get_canteen_data' tool logic: processes input dates/period, fetches data from external API, formats and returns canteen attendance statistics.
@mcp.tool() async def get_canteen_data(start_date: str = None, end_date: str = None, period: str = None) -> str: """获取餐厅就餐人数数据""" valid_periods = ['today', 'yesterday', 'day_before_yesterday', 'this_week', 'last_week', 'this_month', 'last_month'] if period and period in valid_periods: start_date, end_date = get_relative_dates(period) elif period: logger.warning(f"不支持的时间范围: {period}，将使用指定的日期范围") if not start_date or not end_date: # 如果没有指定日期范围，则使用今天 start_date, end_date = get_relative_dates('today') elif not start_date or not end_date: # 如果没有提供日期参数，默认查询今天的数据 start_date, end_date = get_relative_dates('today') # 转换日期格式 try: start_date = convert_date_format(start_date) end_date = convert_date_format(end_date) except ValueError as e: raise ValueError(f"日期格式转换失败: {str(e)}") # 验证转换后的日期格式 if not all(validate_date(date) for date in [start_date, end_date]): raise ValueError("日期格式不正确，请使用YYYYMMDD格式") url = f"{API_BASE}/rsdata/totalnumberofconsumers?startTime={start_date}&endTime={end_date}" try: data = await make_api_request(url) if not data or "success" not in data or not data["success"]: error_msg = data.get("error", "未知错误") if isinstance(data, dict) else "请求失败" raise CanteenAPIError(f"API返回错误: {error_msg}") morning_count = data["data"]["morningCount"] afternoon_count = data["data"]["afternoonCount"] total_count = morning_count + afternoon_count period_titles = { 'today': '今日', 'yesterday': '昨日', 'day_before_yesterday': '前天', 'this_week': '本周', 'last_week': '上周', 'this_month': '本月', 'last_month': '上月' } title = period_titles.get(period, f"{start_date} 至 {end_date}") return f""" 餐厅就餐人数统计 ({title}): 日期范围: {start_date} 至 {end_date} 早餐人数: {morning_count} 人 午餐人数: {afternoon_count} 人 总计: {total_count} 人 """ except CanteenAPIError as e: logger.error(f"获取餐厅数据失败: {str(e)}") raise except Exception as e: logger.error(f"处理餐厅数据时发生错误: {str(e)}") raise CanteenAPIError(f"处理数据失败: {str(e)}") - src/mcp_server_canteen/server.py:176-176 (registration)FastMCP decorator registering the get_canteen_data function as a tool.
@mcp.tool() - Helper function to calculate start and end dates based on relative periods like 'today', 'this_week', etc.
def get_relative_dates(period: str) -> Tuple[str, str]: """获取相对时间范围的开始和结束日期""" today = datetime.now() if period == 'today': return (today.strftime('%Y%m%d'), today.strftime('%Y%m%d')) elif period == 'yesterday': yesterday = today - timedelta(days=1) return (yesterday.strftime('%Y%m%d'), yesterday.strftime('%Y%m%d')) elif period == 'day_before_yesterday': day_before_yesterday = today - timedelta(days=2) return (day_before_yesterday.strftime('%Y%m%d'), day_before_yesterday.strftime('%Y%m%d')) elif period == 'this_week': monday = today - timedelta(days=today.weekday()) return (monday.strftime('%Y%m%d'), today.strftime('%Y%m%d')) elif period == 'last_week': last_monday = today - timedelta(days=today.weekday() + 7) last_sunday = last_monday + timedelta(days=6) return (last_monday.strftime('%Y%m%d'), last_sunday.strftime('%Y%m%d')) elif period == 'this_month': first_day = today.replace(day=1) return (first_day.strftime('%Y%m%d'), today.strftime('%Y%m%d')) elif period == 'last_month': first_day_last_month = (today.replace(day=1) - timedelta(days=1)).replace(day=1) last_day_last_month = today.replace(day=1) - timedelta(days=1) return (first_day_last_month.strftime('%Y%m%d'), last_day_last_month.strftime('%Y%m%d')) else: logger.warning(f"不支持的时间范围: {period}，将使用今天的数据") return (today.strftime('%Y%m%d'), today.strftime('%Y%m%d')) - Helper function to make authenticated HTTP GET requests to the canteen API with error handling.
async def make_api_request(url: str) -> Dict[str, Any]: """向API发送请求并处理错误""" headers = { "Authorization": AUTH_TOKEN, "Accept": "application/json" } try: async with httpx.AsyncClient(timeout=TIMEOUT) as client: response = await client.get(url, headers=headers) response.raise_for_status() return response.json() except httpx.HTTPError as e: logger.error(f"HTTP请求错误: {str(e)}") raise CanteenAPIError(f"HTTP请求失败: {str(e)}") except Exception as e: logger.error(f"未知错误: {str(e)}") raise CanteenAPIError(f"请求失败: {str(e)}") - Helper function to parse and convert various natural language and formatted date strings to standardized YYYYMMDD format.
def convert_date_format(date_str: str) -> str: """将各种日期格式转换为YYYYMMDD格式""" if not date_str: return None # 如果已经是YYYYMMDD格式，直接返回 if len(date_str) == 8 and date_str.isdigit(): return date_str # 处理相对日期（只处理简单的相对日期） today = datetime.now() if date_str.lower() == 'today': return today.strftime('%Y%m%d') elif date_str.lower() == 'yesterday': yesterday = today - timedelta(days=1) return yesterday.strftime('%Y%m%d') elif date_str.lower() == 'day_before_yesterday': day_before_yesterday = today - timedelta(days=2) return day_before_yesterday.strftime('%Y%m%d') # 清理输入字符串 date_str = date_str.strip() # 处理"号"的情况 if "号" in date_str: date_str = date_str.replace("号", "日") # 尝试不同的日期格式 date_formats = [ # 完整日期格式 "%Y-%m-%d", # 2024-04-01 "%Y/%m/%d", # 2024/04/01 "%Y.%m.%d", # 2024.04.01 "%Y年%m月%d日", # 2024年04月01日 # 无年份格式 "%m月%d日", # 04月01日 "%m-%d", # 04-01 "%m/%d", # 04/01 "%m.%d", # 04.01 # 年月格式 "%Y-%m", # 2024-04 "%Y/%m", # 2024/04 "%Y年%m月", # 2024年04月 ] current_year = datetime.now().year for fmt in date_formats: try: date = datetime.strptime(date_str, fmt) # 如果是无年份格式，添加当前年份 if fmt in ["%m月%d日", "%m-%d", "%m/%d", "%m.%d"]: date = date.replace(year=current_year) # 如果是年月格式，添加当月第一天 elif fmt in ["%Y-%m", "%Y/%m", "%Y年%m月"]: date = date.replace(day=1) return date.strftime("%Y%m%d") except ValueError: continue # 如果所有格式都失败，尝试直接解析数字 try: # 移除所有非数字字符 clean_date = ''.join(filter(str.isdigit, date_str)) if len(clean_date) == 8: # 验证日期是否有效 datetime.strptime(clean_date, "%Y%m%d") return clean_date except ValueError: pass raise ValueError(f"无法识别的日期格式: {date_str}")