nat.plugins.mcp.server.memory_profiler#

Memory profiling utilities for MCP frontend.

Attributes#

Classes#

MemoryProfiler

Memory profiler for tracking memory usage and potential leaks.

Module Contents#

logger#
class MemoryProfiler(
enabled: bool = False,
log_interval: int = 50,
top_n: int = 10,
log_level: str = 'DEBUG',
)#

Memory profiler for tracking memory usage and potential leaks.

Initialize the memory profiler.

Args:

enabled: Whether memory profiling is enabled log_interval: Log stats every N requests top_n: Number of top allocations to log log_level: Log level for memory profiling output (e.g., “DEBUG”, “INFO”)

enabled = False#
log_interval#
top_n = 10#
log_level = 10#
request_count = 0#
baseline_snapshot = None#
_we_started_tracemalloc = False#
_log(message: str, *args: Any) None#

Log a message at the configured log level.

Args:

message: Log message format string args: Arguments for the format string

on_request_complete() None#

Called after each request completes.

_ensure_tracing() bool#

Ensure tracemalloc is running if we started it originally.

Returns:

True if tracemalloc is active, False otherwise

_safe_traced_memory() tuple[float, float] | None#

Return (current, peak usage in MB) if tracemalloc is active, else None.

_safe_snapshot() tracemalloc.Snapshot | None#

Return a tracemalloc Snapshot if available, else None.

log_memory_stats() dict[str, Any]#

Log current memory statistics and return them.

_count_instances_of_type(type_name: str) int#

Count instances of a specific type in memory.

_safe_exporter_count() int#
_safe_isolated_exporter_count() int#
_safe_intermediate_step_manager_count() int#
_safe_outstanding_step_count() int#

Get total outstanding steps across all active IntermediateStepManager instances.

get_stats() dict[str, Any]#

Get current memory statistics without logging.

reset_baseline() None#

Reset the baseline snapshot to current state.