mcp-optimizer

Instructions

Implementation Reference

• src/mcp_optimizer/tools/assignment.py:300-332 (handler)

  The primary handler function for the 'solve_assignment_problem_tool' MCP tool. It defines the tool interface, documentation (serving as schema), and delegates to the core solver function.
  @mcp.tool() def solve_assignment_problem_tool( workers: list[str], tasks: list[str], costs: list[list[float]], maximize: bool = False, max_tasks_per_worker: int | None = None, min_tasks_per_worker: int | None = None, ) -> dict[str, Any]: """ Solve assignment problem using OR-Tools Hungarian algorithm. Args: workers: List of worker names tasks: List of task names costs: 2D cost matrix where costs[i][j] is cost of assigning worker i to task j maximize: Whether to maximize instead of minimize (default: False) max_tasks_per_worker: Maximum tasks per worker (optional) min_tasks_per_worker: Minimum tasks per worker (optional) Returns: Dictionary with solution status, assignments, total cost, and execution time """ result = solve_assignment_problem( workers=workers, tasks=tasks, costs=costs, maximize=maximize, max_tasks_per_worker=max_tasks_per_worker, min_tasks_per_worker=min_tasks_per_worker, ) result_dict: dict[str, Any] = result return result_dict
• src/mcp_optimizer/tools/assignment.py:14-148 (helper)

  Core helper function implementing the assignment problem solving logic with input validation, OR-Tools solver invocation, and comprehensive error handling.
  @with_resource_limits(timeout_seconds=60.0, estimated_memory_mb=80.0) def solve_assignment_problem( workers: list[str], tasks: list[str], costs: list[list[float]], maximize: bool = False, max_tasks_per_worker: int | None = None, min_tasks_per_worker: int | None = None, objective: str | None = None, constraints: dict[str, Any] | None = None, ) -> dict[str, Any]: """Solve assignment problem using OR-Tools.""" try: # Validate input if not workers: return { "status": "error", "total_cost": None, "assignments": [], "execution_time": 0.0, "error_message": "No workers provided", } if not tasks: return { "status": "error", "total_cost": None, "assignments": [], "execution_time": 0.0, "error_message": "No tasks provided", } if len(costs) != len(workers): return { "status": "error", "total_cost": None, "assignments": [], "execution_time": 0.0, "error_message": f"Cost matrix dimensions: rows ({len(costs)}) must match workers count ({len(workers)})", } for i, row in enumerate(costs): if len(row) != len(tasks): return { "status": "error", "total_cost": None, "assignments": [], "execution_time": 0.0, "error_message": f"Cost matrix row {i} length ({len(row)}) must match tasks count ({len(tasks)})", } # Handle objective parameter (convert to maximize flag) if objective is not None: maximize = objective.lower() == "maximize" # Handle constraints parameter if constraints is not None: # Check for unsupported constraint types forbidden_assignments = constraints.get("forbidden_assignments", []) task_requirements = constraints.get("task_requirements", {}) if forbidden_assignments: return { "status": "error", "total_cost": None, "assignments": [], "execution_time": 0.0, "error_message": "Forbidden assignments constraints are not currently supported", } if task_requirements: return { "status": "error", "total_cost": None, "assignments": [], "execution_time": 0.0, "error_message": "Task requirements constraints are not currently supported", } # Extract worker limits if provided worker_limits = constraints.get("worker_limits", {}) if worker_limits: # Convert worker limits to max_tasks_per_worker if all workers have same limit limit_values = list(worker_limits.values()) if len(set(limit_values)) == 1 and all( worker in worker_limits for worker in workers ): max_tasks_per_worker = limit_values[0] # If all workers have limit 0 and there are tasks, problem is infeasible if max_tasks_per_worker == 0 and tasks: return { "status": "infeasible", "total_cost": None, "assignments": [], "execution_time": 0.0, "error_message": "No worker can be assigned any tasks due to constraints", } else: return { "status": "error", "total_cost": None, "assignments": [], "execution_time": 0.0, "error_message": "Individual worker limits with different values are not currently supported", } # Create solver and solve from mcp_optimizer.solvers import ORToolsSolver solver = ORToolsSolver() result = solver.solve_assignment_problem( workers=workers, tasks=tasks, costs=costs, maximize=maximize, max_tasks_per_worker=max_tasks_per_worker, min_tasks_per_worker=min_tasks_per_worker, ) # Add objective to result if specified if objective is not None: result["objective"] = objective logger.info(f"Assignment problem solved with status: {result.get('status')}") return result except Exception as e: logger.error(f"Error in solve_assignment_problem: {e}") return { "status": "error", "total_cost": None, "assignments": [], "execution_time": 0.0, "error_message": f"Failed to solve assignment problem: {str(e)}", }
• src/mcp_optimizer/tools/assignment.py:297-357 (registration)

  Registration function that defines and registers the solve_assignment_problem_tool using the @mcp.tool() decorator.
  def register_assignment_tools(mcp: FastMCP[Any]) -> None: """Register assignment and transportation problem tools.""" @mcp.tool() def solve_assignment_problem_tool( workers: list[str], tasks: list[str], costs: list[list[float]], maximize: bool = False, max_tasks_per_worker: int | None = None, min_tasks_per_worker: int | None = None, ) -> dict[str, Any]: """ Solve assignment problem using OR-Tools Hungarian algorithm. Args: workers: List of worker names tasks: List of task names costs: 2D cost matrix where costs[i][j] is cost of assigning worker i to task j maximize: Whether to maximize instead of minimize (default: False) max_tasks_per_worker: Maximum tasks per worker (optional) min_tasks_per_worker: Minimum tasks per worker (optional) Returns: Dictionary with solution status, assignments, total cost, and execution time """ result = solve_assignment_problem( workers=workers, tasks=tasks, costs=costs, maximize=maximize, max_tasks_per_worker=max_tasks_per_worker, min_tasks_per_worker=min_tasks_per_worker, ) result_dict: dict[str, Any] = result return result_dict @mcp.tool() def solve_transportation_problem_tool( suppliers: list[dict[str, Any]], consumers: list[dict[str, Any]], costs: list[list[float]], ) -> dict[str, Any]: """ Solve transportation problem using OR-Tools. Args: suppliers: List of supplier dictionaries with 'name' and 'supply' keys consumers: List of consumer dictionaries with 'name' and 'demand' keys costs: 2D cost matrix where costs[i][j] is cost of shipping from supplier i to consumer j Returns: Dictionary with solution status, flows, total cost, and execution time """ return solve_transportation_problem( suppliers=suppliers, consumers=consumers, costs=costs, ) logger.info("Registered assignment and transportation tools")
• src/mcp_optimizer/mcp_server.py:11-139 (registration)

  Top-level import and call to register_assignment_tools(mcp) in the main MCP server creation, making the tool available.
  from mcp_optimizer.tools.assignment import register_assignment_tools from mcp_optimizer.tools.financial import register_financial_tools from mcp_optimizer.tools.integer_programming import register_integer_programming_tools from mcp_optimizer.tools.knapsack import register_knapsack_tools from mcp_optimizer.tools.linear_programming import register_linear_programming_tools from mcp_optimizer.tools.production import register_production_tools from mcp_optimizer.tools.routing import register_routing_tools from mcp_optimizer.tools.scheduling import register_scheduling_tools from mcp_optimizer.tools.validation import register_validation_tools from mcp_optimizer.utils.resource_monitor import ( get_resource_status, reset_resource_stats, resource_monitor, ) logger = logging.getLogger(__name__) # Server start time for uptime calculation _server_start_time = time.time() def get_health() -> dict[str, Any]: """Get server health status and resource information.""" try: resource_status = get_resource_status() status = "healthy" messages = [] # Check memory usage current_memory = resource_status.get("current_memory_mb", 0) max_memory = resource_status.get("max_memory_mb", 1024) memory_usage_pct = (current_memory / max_memory) * 100 if max_memory > 0 else 0 if memory_usage_pct > 90: status = "critical" messages.append(f"High memory usage: {memory_usage_pct:.1f}%") elif memory_usage_pct > 75: status = "warning" messages.append(f"Elevated memory usage: {memory_usage_pct:.1f}%") # Check active requests active_requests = resource_status.get("active_requests", 0) max_requests = resource_status.get("max_concurrent_requests", 10) if active_requests >= max_requests: status = "warning" messages.append("At maximum concurrent request limit") health_info = { "status": status, "version": get_version("mcp-optimizer"), "uptime": time.time() - _server_start_time, "requests_processed": resource_monitor.total_requests, "resource_stats": resource_monitor.get_stats(), } return { "status": status, "version": get_version("mcp-optimizer"), "uptime": time.time() - _server_start_time, "message": "; ".join(messages), "resource_status": resource_status, "health_info": health_info, } except ImportError as e: return { "status": "error", "message": f"Health check failed: {e}", "resource_status": {}, } def get_resource_stats() -> dict[str, Any]: """Get detailed resource usage statistics.""" return get_resource_status() def reset_resource_statistics() -> dict[str, str]: """Reset resource monitoring statistics.""" reset_resource_stats() return {"status": "reset", "message": "Resource statistics have been reset"} def get_server_info() -> dict[str, Any]: """Get comprehensive server information.""" return { "name": "MCP Optimizer", "version": get_version("mcp-optimizer"), "description": "Mathematical optimization server with multiple solvers", "uptime": time.time() - _server_start_time, "capabilities": { "linear_programming": True, "integer_programming": True, "mixed_integer_programming": True, "assignment_problems": True, "transportation_problems": True, "knapsack_problems": True, "routing_problems": True, "scheduling_problems": True, "portfolio_optimization": True, "production_planning": True, "input_validation": True, }, "solvers": { "pulp": "Linear/Integer Programming", "ortools": "Routing, Scheduling, Assignment", "native": "Portfolio, Production Planning", }, "configuration": { "max_solve_time": settings.max_solve_time, "max_memory_mb": settings.max_memory_mb, "max_concurrent_requests": settings.max_concurrent_requests, "log_level": settings.log_level.value, "debug": settings.debug, }, } def create_mcp_server() -> FastMCP[dict[str, str]]: """Create and configure the MCP server with optimization tools.""" # Create MCP server mcp: FastMCP[dict[str, str]] = FastMCP("MCP Optimizer") # Register all optimization tools register_linear_programming_tools(mcp) register_integer_programming_tools(mcp) register_assignment_tools(mcp)