mcp-optimizer

Instructions

Implementation Reference

• src/mcp_optimizer/tools/scheduling.py:547-577 (handler)

  The decorated MCP tool handler that accepts parameters, constructs input data, invokes the shift scheduling solver, and returns the result.
  @mcp.tool() def solve_employee_shift_scheduling( employees: list[str], shifts: list[dict[str, Any]], days: int, employee_constraints: dict[str, dict[str, Any]] | None = None, time_limit_seconds: float = 30.0, ) -> dict[str, Any]: """Solve Employee Shift Scheduling to assign employees to shifts optimally. Args: employees: List of employee names shifts: List of shift dictionaries with time and requirements days: Number of days to schedule employee_constraints: Optional constraints and preferences per employee time_limit_seconds: Maximum solving time in seconds (default: 30.0) Returns: Optimization result with employee schedules and coverage statistics """ input_data = { "employees": employees, "shifts": shifts, "days": days, "employee_constraints": employee_constraints or {}, "time_limit_seconds": time_limit_seconds, } result = solve_shift_scheduling(input_data) result_dict: dict[str, Any] = result.model_dump() return result_dict
• src/mcp_optimizer/tools/scheduling.py:109-131 (schema)

  Pydantic model defining and validating the input structure for the shift scheduling solver.
  class ShiftSchedulingInput(BaseModel): """Input schema for Shift Scheduling.""" employees: list[str] shifts: list[Shift] days: int = Field(ge=1) employee_constraints: dict[str, EmployeeConstraints] = Field(default_factory=dict) time_limit_seconds: float = Field(default=30.0, ge=0) @field_validator("employees") @classmethod def validate_employees(cls, v: list[str]) -> list[str]: if not v: raise ValueError("Must have at least one employee") return v @field_validator("shifts") @classmethod def validate_shifts(cls, v: list[Shift]) -> list[Shift]: if not v: raise ValueError("Must have at least one shift") return v
• src/mcp_optimizer/mcp_server.py:142-142 (registration)

  Invocation of register_scheduling_tools during MCP server setup, which defines and registers the solve_employee_shift_scheduling tool.
  register_scheduling_tools(mcp)
• src/mcp_optimizer/tools/scheduling.py:308-509 (helper)

  Core solver function implementing the OR-Tools CP-SAT model for employee shift assignment optimization, handling constraints and objectives.
  @with_resource_limits(timeout_seconds=90.0, estimated_memory_mb=120.0) def solve_shift_scheduling(input_data: dict[str, Any]) -> OptimizationResult: """Solve Shift Scheduling Problem using OR-Tools CP-SAT. Args: input_data: Shift scheduling problem specification Returns: OptimizationResult with employee shift assignments """ if not ORTOOLS_AVAILABLE: return OptimizationResult( status=OptimizationStatus.ERROR, objective_value=None, variables={}, execution_time=0.0, error_message="OR-Tools is not available. Please install it with 'pip install ortools'", ) start_time = time.time() try: # Parse and validate input scheduling_input = ShiftSchedulingInput(**input_data) employees = scheduling_input.employees shifts = scheduling_input.shifts days = scheduling_input.days employee_constraints = scheduling_input.employee_constraints # Create CP-SAT model model = cp_model.CpModel() # Variables: assignment[employee][shift][day] = 1 if employee works shift on day assignments: dict[int, dict[int, dict[int, Any]]] = {} for emp_idx, employee in enumerate(employees): assignments[emp_idx] = {} for shift_idx, shift in enumerate(shifts): assignments[emp_idx][shift_idx] = {} for day in range(days): var_name = f"assign_{employee}_{shift.name}_{day}" assignments[emp_idx][shift_idx][day] = model.NewBoolVar(var_name) # Constraint: Each shift must have required staff each day for shift_idx, shift in enumerate(shifts): for day in range(days): model.Add( sum(assignments[emp_idx][shift_idx][day] for emp_idx in range(len(employees))) >= shift.required_staff ) # Employee constraints for emp_idx, employee in enumerate(employees): emp_constraints = employee_constraints.get(employee, EmployeeConstraints()) # Max/min shifts per week if emp_constraints.max_shifts_per_week is not None: total_shifts = sum( assignments[emp_idx][shift_idx][day] for shift_idx in range(len(shifts)) for day in range(days) ) model.Add(total_shifts <= emp_constraints.max_shifts_per_week) if emp_constraints.min_shifts_per_week is not None: total_shifts = sum( assignments[emp_idx][shift_idx][day] for shift_idx in range(len(shifts)) for day in range(days) ) model.Add(total_shifts >= emp_constraints.min_shifts_per_week) # Unavailable shifts for shift_name in emp_constraints.unavailable_shifts: for shift_idx, shift in enumerate(shifts): if shift.name == shift_name: for day in range(days): model.Add(assignments[emp_idx][shift_idx][day] == 0) # Skills requirements for shift_idx, shift in enumerate(shifts): if shift.skills_required: has_required_skills = all( skill in emp_constraints.skills for skill in shift.skills_required ) if not has_required_skills: for day in range(days): model.Add(assignments[emp_idx][shift_idx][day] == 0) # No overlapping shifts on same day for day in range(days): overlapping_shifts = [] for shift_idx, _shift in enumerate(shifts): overlapping_shifts.append(assignments[emp_idx][shift_idx][day]) model.Add(sum(overlapping_shifts) <= 1) # Max consecutive shifts if emp_constraints.max_consecutive_shifts is not None: for start_day in range(days - emp_constraints.max_consecutive_shifts): consecutive_vars = [] for day in range( start_day, start_day + emp_constraints.max_consecutive_shifts + 1, ): day_working = model.NewBoolVar(f"working_{employee}_{day}") model.Add( day_working == sum( assignments[emp_idx][shift_idx][day] for shift_idx in range(len(shifts)) ) ) consecutive_vars.append(day_working) model.Add(sum(consecutive_vars) <= emp_constraints.max_consecutive_shifts) # Objective: Minimize total assignments (prefer fewer shifts) and maximize preferences total_assignments = sum( assignments[emp_idx][shift_idx][day] for emp_idx in range(len(employees)) for shift_idx in range(len(shifts)) for day in range(days) ) # Add preference bonus preference_bonus = 0 for emp_idx, employee in enumerate(employees): emp_constraints = employee_constraints.get(employee, EmployeeConstraints()) for shift_name in emp_constraints.preferred_shifts: for shift_idx, shift in enumerate(shifts): if shift.name == shift_name: preference_bonus += sum( assignments[emp_idx][shift_idx][day] for day in range(days) ) # Minimize negative preference (maximize preference) model.Minimize(total_assignments - preference_bonus) # Solve solver = cp_model.CpSolver() solver.parameters.max_time_in_seconds = scheduling_input.time_limit_seconds status = solver.Solve(model) if status == cp_model.OPTIMAL or status == cp_model.FEASIBLE: # type: ignore[comparison-overlap,unused-ignore] # Extract solution schedule = [] total_cost = 0 for emp_idx, employee in enumerate(employees): employee_schedule = [] for day in range(days): day_shifts = [] for shift_idx, shift in enumerate(shifts): if solver.Value(assignments[emp_idx][shift_idx][day]): day_shifts.append( { "shift_name": shift.name, "start": shift.start, "end": shift.end, "skills_required": shift.skills_required, } ) total_cost += 1 employee_schedule.append({"day": day, "shifts": day_shifts}) schedule.append({"employee": employee, "schedule": employee_schedule}) execution_time = time.time() - start_time return OptimizationResult( status=OptimizationStatus.OPTIMAL if status == cp_model.OPTIMAL # type: ignore[comparison-overlap,unused-ignore] else OptimizationStatus.FEASIBLE, objective_value=float(total_cost), variables={ "schedule": schedule, "total_assignments": total_cost, "num_employees": len(employees), "num_shifts": len(shifts), "num_days": days, }, execution_time=execution_time, solver_info={ "solver_name": "OR-Tools CP-SAT", "status": solver.StatusName(status), }, ) else: status_name = solver.StatusName(status) return OptimizationResult( status=OptimizationStatus.INFEASIBLE if status == cp_model.INFEASIBLE # type: ignore[comparison-overlap,unused-ignore] else OptimizationStatus.ERROR, error_message=f"No solution found: {status_name}", execution_time=time.time() - start_time, ) except Exception as e: return OptimizationResult( status=OptimizationStatus.ERROR, error_message=f"Shift scheduling error: {str(e)}", execution_time=time.time() - start_time, )