DataBeak

Instructions

Implementation Reference

• src/databeak/servers/system_server.py:83-182 (handler)

  The core handler function implementing the health_check MCP tool. It assesses server health including session count, memory usage via helpers, determines status (healthy/degraded/unhealthy), and returns a structured HealthResult with progress updates via context.
  async def health_check( ctx: Annotated[Context, Field(description="FastMCP context for progress reporting")], ) -> HealthResult: """Check DataBeak server health and availability with memory monitoring. Returns server status, session capacity, memory usage, and version information. Use before large operations to verify system readiness and resource availability. """ try: await ctx.info("Performing DataBeak health check with memory monitoring") session_manager = get_session_manager() settings = get_settings() active_sessions = len(session_manager.sessions) # Get memory information current_memory_mb = get_memory_usage() health_threshold_mb = float(settings.health_memory_threshold_mb) memory_status = get_memory_status(current_memory_mb, health_threshold_mb) # Determine overall health status status = "healthy" # Check session capacity using configurable threshold session_capacity_ratio = active_sessions / session_manager.max_sessions if session_capacity_ratio >= settings.session_capacity_warning_threshold: status = "degraded" await ctx.warning( f"Session capacity warning: {active_sessions}/{session_manager.max_sessions} " f"({session_capacity_ratio:.1%})" ) # Check memory status if memory_status == "critical": status = "unhealthy" await ctx.error( f"Critical memory usage: {current_memory_mb:.1f}MB / {health_threshold_mb:.1f}MB" ) elif memory_status == "warning": if status == "healthy": status = "degraded" await ctx.warning( f"High memory usage: {current_memory_mb:.1f}MB / {health_threshold_mb:.1f}MB" ) await ctx.info( f"Health check complete - Status: {status}, Sessions: {active_sessions}, " f"Memory: {current_memory_mb:.1f}MB ({memory_status})" ) return HealthResult( status=status, version=__version__, active_sessions=active_sessions, max_sessions=session_manager.max_sessions, session_ttl_minutes=session_manager.ttl_minutes, memory_usage_mb=current_memory_mb, memory_threshold_mb=health_threshold_mb, memory_status=memory_status, history_operations_total=0, # History operations tracking removed history_limit_per_session=0, # History operations tracking removed ) except (ImportError, AttributeError, ValueError, TypeError) as e: # Handle specific configuration/import issues - return unhealthy await ctx.error(f"Health check failed due to configuration issue: {e}") return HealthResult( status="unhealthy", version="unknown", active_sessions=0, max_sessions=0, session_ttl_minutes=0, memory_usage_mb=0.0, memory_threshold_mb=2048.0, memory_status="unknown", history_operations_total=0, # History operations tracking removed history_limit_per_session=0, # History operations tracking removed ) except Exception as e: # Treat unexpected session manager errors as recoverable - return unhealthy await ctx.error(f"Health check failed: {e}") try: version = str(__version__) except Exception: version = "unknown" return HealthResult( status="unhealthy", version=version, active_sessions=0, max_sessions=0, session_ttl_minutes=0, memory_usage_mb=0.0, memory_threshold_mb=2048.0, memory_status="unknown", history_operations_total=0, # History operations tracking removed history_limit_per_session=0, # History operations tracking removed )
• src/databeak/models/tool_responses.py:61-74 (schema)

  Pydantic model defining the structured output schema for the health_check tool, including status, version, session info, and memory metrics.
  class HealthResult(BaseToolResponse): """Response model for system health check with memory monitoring.""" status: str = Field(description="Server health status: healthy, degraded, or unhealthy") version: str = Field(description="DataBeak server version") active_sessions: int = Field(description="Number of currently active data sessions") max_sessions: int = Field(description="Maximum allowed concurrent sessions") session_ttl_minutes: int = Field(description="Session timeout in minutes") memory_usage_mb: float = Field(description="Current memory usage in MB") memory_threshold_mb: float = Field(description="Memory usage threshold in MB") memory_status: str = Field(description="Memory status: normal, warning, critical") history_operations_total: int = Field(description="Total operations in all session histories") history_limit_per_session: int = Field(description="Maximum operations per session history")
• src/databeak/servers/system_server.py:268-270 (registration)

  Direct registration of the health_check function as an MCP tool on the system_server FastMCP instance.
  # Register the system functions directly as MCP tools (no wrapper functions needed) system_server.tool(name="health_check")(health_check) system_server.tool(name="get_server_info")(get_server_info)
• src/databeak/server.py:103-113 (registration)

  Mounting of the system_server (containing health_check tool) onto the main DataBeak FastMCP server instance in create_server().
  # Mount specialized servers mcp.mount(system_server) mcp.mount(io_server) mcp.mount(row_operations_server) mcp.mount(statistics_server) mcp.mount(discovery_server) mcp.mount(validation_server) mcp.mount(transformation_server) mcp.mount(column_server) mcp.mount(column_text_server)
• src/databeak/servers/system_server.py:33-50 (helper)

  Helper function to retrieve current process memory usage in MB, used by health_check.
  def get_memory_usage() -> float: """Get current process memory usage in MB. Returns: Memory usage in MB, or 0.0 if measurement fails Raises: None - errors are logged and 0.0 is returned for resilience """ try: process = psutil.Process(os.getpid()) memory_bytes: int = process.memory_info().rss return float(memory_bytes / 1024 / 1024) except (psutil.Error, OSError) as e: logger.warning("Failed to get memory usage: %s", str(e)) return 0.0