mcp-optimizer

Instructions

Implementation Reference

• src/mcp_optimizer/tools/scheduling.py:515-546 (handler)

  The main handler function for the 'solve_job_shop_scheduling' tool, registered with @mcp.tool(). It prepares the input dictionary and calls the core solver function.
  @mcp.tool() def solve_job_shop_scheduling( jobs: list[dict[str, Any]], machines: list[str], horizon: int, objective: str = "makespan", time_limit_seconds: float = 30.0, ) -> dict[str, Any]: """Solve Job Shop Scheduling Problem to optimize machine utilization and completion times. Args: jobs: List of job dictionaries with tasks and constraints machines: List of available machine names horizon: Maximum time horizon for scheduling objective: Optimization objective ("makespan" or "total_completion_time") time_limit_seconds: Maximum solving time in seconds (default: 30.0) Returns: Optimization result with job schedule and machine assignments """ input_data = { "jobs": jobs, "machines": machines, "horizon": horizon, "objective": objective, "time_limit_seconds": time_limit_seconds, } result = solve_job_scheduling(input_data) result_dict: dict[str, Any] = result.model_dump() return result_dict
• src/mcp_optimizer/tools/scheduling.py:57-79 (schema)

  Pydantic schema for validating input to the job shop scheduling problem, including jobs, machines, horizon, objective, and time limit.
  class JobSchedulingInput(BaseModel): """Input schema for Job Shop Scheduling.""" jobs: list[Job] machines: list[str] horizon: int = Field(ge=1) objective: str = Field(default="makespan", pattern="^(makespan|total_completion_time)$") time_limit_seconds: float = Field(default=30.0, ge=0) @field_validator("jobs") @classmethod def validate_jobs(cls, v: list[Job]) -> list[Job]: if not v: raise ValueError("Must have at least one job") return v @field_validator("machines") @classmethod def validate_machines(cls, v: list[str]) -> list[str]: if not v: raise ValueError("Must have at least one machine") return v
• src/mcp_optimizer/tools/scheduling.py:133-305 (helper)

  Core helper function that implements the Job Shop Scheduling optimization logic using OR-Tools CP-SAT solver, handling variables, constraints, objectives, and extracting the schedule.
  @with_resource_limits(timeout_seconds=120.0, estimated_memory_mb=150.0) def solve_job_scheduling(input_data: dict[str, Any]) -> OptimizationResult: """Solve Job Shop Scheduling Problem using OR-Tools CP-SAT. Args: input_data: Job scheduling problem specification Returns: OptimizationResult with job schedule and makespan """ 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 = JobSchedulingInput(**input_data) jobs = scheduling_input.jobs machines = scheduling_input.machines horizon = scheduling_input.horizon # Create CP-SAT model model = cp_model.CpModel() # Variables for each task: (start_time, end_time, interval) task_vars: dict[tuple, tuple] = {} machine_intervals: dict[int, list] = {i: [] for i in range(len(machines))} # Create variables for each task for job in jobs: for task_idx, task in enumerate(job.tasks): suffix = f"_{job.id}_{task_idx}" start_var = model.NewIntVar(0, horizon, f"start{suffix}") duration = task.duration + task.setup_time end_var = model.NewIntVar(0, horizon, f"end{suffix}") interval_var = model.NewIntervalVar( start_var, duration, end_var, f"interval{suffix}" ) task_vars[(job.id, task_idx)] = (start_var, end_var, interval_var) machine_intervals[task.machine].append(interval_var) # Add precedence constraints within jobs for job in jobs: for task_idx in range(len(job.tasks) - 1): _, end_var, _ = task_vars[(job.id, task_idx)] start_var_next, _, _ = task_vars[(job.id, task_idx + 1)] model.Add(end_var <= start_var_next) # Add machine capacity constraints (no overlap) for machine_idx in range(len(machines)): if machine_intervals[machine_idx]: model.AddNoOverlap(machine_intervals[machine_idx]) # Add release time constraints for job in jobs: if job.release_time > 0: start_var, _, _ = task_vars[(job.id, 0)] model.Add(start_var >= job.release_time) # Add deadline constraints for job in jobs: if job.deadline is not None: _, end_var, _ = task_vars[(job.id, len(job.tasks) - 1)] model.Add(end_var <= job.deadline) # Objective: minimize makespan or total completion time if scheduling_input.objective == "makespan": makespan = model.NewIntVar(0, horizon, "makespan") for job in jobs: _, end_var, _ = task_vars[(job.id, len(job.tasks) - 1)] model.Add(makespan >= end_var) model.Minimize(makespan) else: # total_completion_time completion_times = [] for job in jobs: _, end_var, _ = task_vars[(job.id, len(job.tasks) - 1)] completion_times.append(end_var) model.Minimize(sum(completion_times)) # 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 = [] job_completion_times = {} for job in jobs: job_schedule = [] for task_idx, task in enumerate(job.tasks): start_var, end_var, _ = task_vars[(job.id, task_idx)] start_time_val = solver.Value(start_var) end_time_val = solver.Value(end_var) job_schedule.append( { "task_index": task_idx, "machine": machines[task.machine], "machine_index": task.machine, "start_time": start_time_val, "end_time": end_time_val, "duration": task.duration, "setup_time": task.setup_time, } ) completion_time = max(task["end_time"] for task in job_schedule) # type: ignore[type-var] job_completion_times[job.id] = completion_time schedule.append( { "job_id": job.id, "tasks": job_schedule, "completion_time": completion_time, "priority": job.priority, } ) makespan = max(job_completion_times.values()) if job_completion_times else 0 # type: ignore[type-var,assignment] total_completion_time = sum(job_completion_times.values()) # type: ignore[arg-type] 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( makespan # type: ignore[arg-type] if scheduling_input.objective == "makespan" else total_completion_time ), variables={ "schedule": schedule, "makespan": makespan, "total_completion_time": total_completion_time, "job_completion_times": job_completion_times, "num_jobs": len(jobs), "num_machines": len(machines), }, execution_time=execution_time, solver_info={ "solver_name": "OR-Tools CP-SAT", "status": solver.StatusName(status), "objective": scheduling_input.objective, }, ) 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"Job scheduling error: {str(e)}", execution_time=time.time() - start_time, )
• src/mcp_optimizer/mcp_server.py:137-145 (registration)

  Registration of all tools including scheduling tools via register_scheduling_tools(mcp) in the MCP server creation function.
  register_linear_programming_tools(mcp) register_integer_programming_tools(mcp) register_assignment_tools(mcp) register_knapsack_tools(mcp) register_routing_tools(mcp) register_scheduling_tools(mcp) register_financial_tools(mcp) register_production_tools(mcp) register_validation_tools(mcp)
• src/mcp_optimizer/tools/scheduling.py:512-579 (registration)

  Function that registers the scheduling tools, including 'solve_job_shop_scheduling', by defining them with @mcp.tool() decorators.
  def register_scheduling_tools(mcp: FastMCP[Any]) -> None: """Register scheduling optimization tools with MCP server.""" @mcp.tool() def solve_job_shop_scheduling( jobs: list[dict[str, Any]], machines: list[str], horizon: int, objective: str = "makespan", time_limit_seconds: float = 30.0, ) -> dict[str, Any]: """Solve Job Shop Scheduling Problem to optimize machine utilization and completion times. Args: jobs: List of job dictionaries with tasks and constraints machines: List of available machine names horizon: Maximum time horizon for scheduling objective: Optimization objective ("makespan" or "total_completion_time") time_limit_seconds: Maximum solving time in seconds (default: 30.0) Returns: Optimization result with job schedule and machine assignments """ input_data = { "jobs": jobs, "machines": machines, "horizon": horizon, "objective": objective, "time_limit_seconds": time_limit_seconds, } result = solve_job_scheduling(input_data) result_dict: dict[str, Any] = result.model_dump() return result_dict @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 logger.info("Registered scheduling tools")