feat: Introduce PyEval functionality for secure Python expression evaluation (#470)

- Added a new endpoint for evaluating Python expressions via the PyEval API.
- Implemented a dedicated page in the web interface for users to input and evaluate expressions safely.
- Integrated the RestrictedPythonEvaluator to ensure secure execution of user-provided code.
- Updated the main application routes and templates to include navigation to the new PyEval feature.
- Enhanced the evaluator with additional safe built-in functions and improved error handling.

This update significantly expands the capabilities of the application, allowing users to safely evaluate Python expressions in a controlled environment.

Author: plusplusoneplusplus

server/api/__init__.py

@@ -67,6 +67,8 @@
     api_export_dataframe,
 )
 
+from .pyeval import pyeval_routes
+
 # Aggregate all routes into a single list
 api_routes = [
     # Knowledge management endpoints
@@ -127,4 +129,4 @@
     Route("/api/dataframes/{df_id}/summary", endpoint=api_get_dataframe_summary, methods=["GET"]),
     Route("/api/dataframes/{df_id}/execute", endpoint=api_execute_dataframe_operation, methods=["POST"]),
     Route("/api/dataframes/{df_id}/export", endpoint=api_export_dataframe, methods=["POST"]),
-]
+] + pyeval_routes

server/api/pyeval.py

@@ -0,0 +1,141 @@
+"""API endpoints for Python expression evaluation using the pyeval utility."""
+
+import logging
+from typing import Any, Dict, Optional
+
+from starlette.requests import Request
+from starlette.responses import JSONResponse
+from starlette.routing import Route
+
+from utils.pyeval.evaluator import RestrictedPythonEvaluator, EvaluationError
+
+logger = logging.getLogger(__name__)
+
+
+async def evaluate_expression(request: Request) -> JSONResponse:
+    """Evaluate a Python expression using the RestrictedPythonEvaluator.
+
+    POST /api/pyeval/evaluate
+
+    Request body:
+    {
+        "expression": "python_expression_to_evaluate",
+        "context": {
+            "variable_name": "variable_value",
+            ...
+        }
+    }
+
+    Response:
+    {
+        "success": true/false,
+        "result": "evaluation_result" | null,
+        "execution_time_ms": 123.45,
+        "error_message": "error_description" | null
+    }
+    """
+    try:
+        # Parse request body
+        body = await request.json()
+        expression = body.get("expression", "").strip()
+        context = body.get("context", {})
+
+        # Validate input
+        if not expression:
+            return JSONResponse({
+                "success": False,
+                "result": None,
+                "execution_time_ms": 0.0,
+                "error_message": "Expression cannot be empty"
+            }, status_code=400)
+
+        if not isinstance(context, dict):
+            return JSONResponse({
+                "success": False,
+                "result": None,
+                "execution_time_ms": 0.0,
+                "error_message": "Context must be a dictionary"
+            }, status_code=400)
+
+        logger.info(f"Evaluating expression: {expression[:100]}{'...' if len(expression) > 100 else ''}")
+        logger.debug(f"Context variables: {list(context.keys())}")
+
+        # Create evaluator and evaluate expression
+        evaluator = RestrictedPythonEvaluator()
+        result = evaluator.evaluate_expression(expression, context)
+
+        # Format result for JSON response
+        if result.success:
+            # Convert result to string representation for JSON serialization
+            result_str = _format_result_for_json(result.result)
+            logger.info(f"Expression evaluated successfully in {result.execution_time_ms:.2f}ms")
+
+            return JSONResponse({
+                "success": True,
+                "result": result_str,
+                "execution_time_ms": result.execution_time_ms,
+                "error_message": None
+            })
+        else:
+            logger.warning(f"Expression evaluation failed: {result.error_message}")
+            return JSONResponse({
+                "success": False,
+                "result": None,
+                "execution_time_ms": result.execution_time_ms,
+                "error_message": result.error_message
+            })
+
+    except Exception as e:
+        logger.error(f"Error in evaluate_expression endpoint: {e}", exc_info=True)
+        return JSONResponse({
+            "success": False,
+            "result": None,
+            "execution_time_ms": 0.0,
+            "error_message": f"Server error: {str(e)}"
+        }, status_code=500)
+
+
+def _format_result_for_json(result: Any) -> str:
+    """Format evaluation result for JSON serialization.
+
+    Args:
+        result: The result from expression evaluation
+
+    Returns:
+        String representation of the result suitable for JSON response
+    """
+    try:
+        # Handle pandas DataFrames specially
+        if hasattr(result, 'to_string'):
+            # This covers pandas DataFrames and Series
+            return result.to_string()
+
+        # Handle numpy arrays
+        elif hasattr(result, 'tolist'):
+            return str(result)
+
+        # Handle other iterables (but not strings)
+        elif hasattr(result, '__iter__') and not isinstance(result, (str, bytes)):
+            # Convert to string representation, but limit length for very large iterables
+            str_result = str(result)
+            if len(str_result) > 10000:
+                return str_result[:10000] + "... (truncated)"
+            return str_result
+
+        # Handle basic types
+        else:
+            str_result = str(result)
+            # Limit very long string results
+            if len(str_result) > 10000:
+                return str_result[:10000] + "... (truncated)"
+            return str_result
+
+    except Exception as e:
+        logger.warning(f"Error formatting result for JSON: {e}")
+        return f"<Error formatting result: {str(e)}>"
+
+
+# API routes for pyeval functionality
+pyeval_routes = [
+    Route("/api/pyeval/evaluate", endpoint=evaluate_expression, methods=["POST"]),
+]

server/main.py

@@ -391,6 +391,11 @@ async def dataframe_detail_page(request: Request):
         "dataframe_detail.html", {"request": request, "current_page": "dataframes", "df_id": df_id}
     )
 
+async def pyeval_page(request: Request):
+    return templates.TemplateResponse(
+        "pyeval.html", {"request": request, "current_page": "pyeval"}
+    )
+
 
 # --- Add routes ---
 routes = [
@@ -402,6 +407,7 @@ async def dataframe_detail_page(request: Request):
     Route("/visualizations", endpoint=visualizations_page, methods=["GET"]),
     Route("/dataframes", endpoint=dataframes_page, methods=["GET"]),
     Route("/dataframes/{df_id}", endpoint=dataframe_detail_page, methods=["GET"]),
+    Route("/pyeval", endpoint=pyeval_page, methods=["GET"]),
     Route("/config", endpoint=config_page, methods=["GET"]),
     Route("/sse", endpoint=handle_sse),
     Mount("/messages/", app=sse.handle_post_message),

server/templates/base.html

@@ -370,6 +370,7 @@ <h1>MCP Knowledge Server</h1>
             <a href="/knowledge" class="nav-link {{ 'active' if current_page == 'knowledge' else '' }}">Knowledge</a>
             <a href="/jobs" class="nav-link {{ 'active' if current_page == 'jobs' else '' }}">Background Jobs</a>
             <a href="/dataframes" class="nav-link {{ 'active' if current_page == 'dataframes' else '' }}">DataFrames</a>
+            <a href="/pyeval" class="nav-link {{ 'active' if current_page == 'pyeval' else '' }}">PyEval</a>
             <a href="/tools" class="nav-link {{ 'active' if current_page == 'tools' else '' }}">Tools</a>
             <a href="/tool-history" class="nav-link {{ 'active' if current_page == 'tool_history' else '' }}">Tool
                 History</a>