get_network_info_tool

Instructions

Implementation Reference

• src/system_info_mcp/server.py:62-65 (registration)

  Registration of the 'get_network_info_tool' MCP tool via FastMCP @app.tool() decorator. Delegates to get_network_info() in tools.py.

  @app.tool()
  def get_network_info_tool() -> Dict[str, Any]:
      """Retrieve network interface information and statistics."""
      return get_network_info()

• src/system_info_mcp/tools.py:179-256 (handler)

  Core handler that uses psutil to retrieve network interface addresses, statistics (is_up, speed, mtu), and I/O counters (bytes/packets sent/received, errors, drops). Results are cached with a 5-second TTL.

  @cache_result("network_info", ttl=5)
  def get_network_info() -> Dict[str, Any]:
      """Retrieve network interface information and statistics."""
      try:
          interfaces = []
  
          # Get network interfaces
          net_if_addrs = psutil.net_if_addrs()
          net_if_stats = psutil.net_if_stats()
  
          for interface_name, addresses in net_if_addrs.items():
              interface_info: Dict[str, Any] = {
                  "name": interface_name,
                  "addresses": [],
                  "is_up": False,
                  "speed": 0,
                  "mtu": 0,
              }
  
              # Get interface statistics
              if interface_name in net_if_stats:
                  stats = net_if_stats[interface_name]
                  interface_info.update(
                      {"is_up": stats.isup, "speed": stats.speed, "mtu": stats.mtu}
                  )
  
              # Get addresses
              for addr in addresses:
                  addr_info = {"family": str(addr.family), "address": addr.address}
                  if addr.netmask:
                      addr_info["netmask"] = addr.netmask
                  interface_info["addresses"].append(addr_info)
  
              interfaces.append(interface_info)
  
          # Get network I/O statistics
          try:
              net_io = psutil.net_io_counters()
              if net_io:
                  io_stats = {
                      "bytes_sent": net_io.bytes_sent,
                      "bytes_recv": net_io.bytes_recv,
                      "packets_sent": net_io.packets_sent,
                      "packets_recv": net_io.packets_recv,
                      "errin": net_io.errin,
                      "errout": net_io.errout,
                      "dropin": net_io.dropin,
                      "dropout": net_io.dropout,
                  }
              else:
                  io_stats = {
                      "bytes_sent": 0,
                      "bytes_recv": 0,
                      "packets_sent": 0,
                      "packets_recv": 0,
                      "errin": 0,
                      "errout": 0,
                      "dropin": 0,
                      "dropout": 0,
                  }
          except Exception as e:
              logger.warning(f"Could not get network I/O stats: {e}")
              io_stats = {
                  "bytes_sent": 0,
                  "bytes_recv": 0,
                  "packets_sent": 0,
                  "packets_recv": 0,
                  "errin": 0,
                  "errout": 0,
                  "dropin": 0,
                  "dropout": 0,
              }
  
          return {"interfaces": interfaces, "stats": io_stats}
  
      except Exception as e:
          logger.error(f"Error getting network info: {e}")
          raise

• src/system_info_mcp/utils.py:55-107 (helper)

  Caching decorator applied to get_network_info with a 5-second TTL. Caches results by key to avoid redundant psutil calls.

  def cache_result(cache_key: str, ttl: Optional[int] = None) -> Any:
      """Decorator to cache function results with TTL."""
      if ttl is None:
          ttl = config.cache_ttl
  
      def decorator(func: Callable[..., T]) -> Callable[..., T]:
          @wraps(func)
          async def async_wrapper(*args: Any, **kwargs: Any) -> T:
              current_time = time.time()
  
              # Check if we have cached result
              if cache_key in _cache:
                  cache_entry = _cache[cache_key]
                  if current_time - cache_entry["timestamp"] < ttl:
                      logger.debug(f"Cache hit for {cache_key}")
                      return cache_entry["data"]
  
              # Get fresh data
              logger.debug(f"Cache miss for {cache_key}, fetching fresh data")
              if asyncio.iscoroutinefunction(func):
                  result = await func(*args, **kwargs)
              else:
                  result = func(*args, **kwargs)
  
              # Cache the result
              _cache[cache_key] = {"data": result, "timestamp": current_time}
  
              return result
  
          @wraps(func)
          def sync_wrapper(*args: Any, **kwargs: Any) -> T:
              current_time = time.time()
  
              # Check if we have cached result
              if cache_key in _cache:
                  cache_entry = _cache[cache_key]
                  if current_time - cache_entry["timestamp"] < ttl:
                      logger.debug(f"Cache hit for {cache_key}")
                      return cache_entry["data"]
  
              # Get fresh data
              logger.debug(f"Cache miss for {cache_key}, fetching fresh data")
              result = func(*args, **kwargs)
  
              # Cache the result
              _cache[cache_key] = {"data": result, "timestamp": current_time}
  
              return result
  
          # Return appropriate wrapper based on whether function is async
          return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper
  
      return decorator