MCP Filesystem Server

Instructions

Implementation Reference

• mcp_filesystem/server.py:806-850 (handler)

  The MCP tool handler function for 'read_file_lines'. Decorated with @mcp.tool(), it handles the tool invocation, calls the core operations method, formats the output with metadata header, and manages errors. This is the entry point called by the MCP server.
  async def read_file_lines( path: str, ctx: Context, offset: int = 0, limit: Optional[int] = None, encoding: str = "utf-8", ) -> str: """Read specific lines from a text file. Args: path: Path to the file offset: Line offset (0-based, starts at first line) limit: Maximum number of lines to read (None for all remaining) encoding: Text encoding (default: utf-8) ctx: MCP context Returns: File content and metadata """ try: components = get_components() content, metadata = await components["operations"].read_file_lines( path, offset, limit, encoding ) if not content: last_line_desc = "end" if limit is None else f"offset+{limit}" return f"No content found between offset {offset} and {last_line_desc}" # Calculate display lines (1-based for human readability) display_start = offset + 1 display_end = offset + metadata["lines_read"] header = ( f"File: {path}\n" f"Lines: {display_start} to {display_end} " f"(of {metadata['total_lines']} total)\n" f"----------------------------------------\n" ) return header + content except Exception as e: return f"Error reading file lines: {str(e)}"
• mcp_filesystem/operations.py:574-691 (helper)

  Core helper method in FileOperations class that implements the logic for reading specific lines from a file efficiently by precomputing line positions. Returns content and detailed metadata. Called by the MCP tool handler.
  async def read_file_lines( self, path: Union[str, Path], offset: int = 0, limit: Optional[int] = None, encoding: str = "utf-8", ) -> Tuple[str, Dict[str, Any]]: """Read specific lines from a text file using offset and limit. Args: path: Path to the file offset: Line offset (0-based, starts at first line) limit: Maximum number of lines to read (None for all remaining) encoding: Text encoding (default: utf-8) Returns: Tuple of (file content, metadata) Raises: ValueError: If path is outside allowed directories FileNotFoundError: If file does not exist """ abs_path, allowed = await self.validator.validate_path(path) if not allowed: raise ValueError(f"Path outside allowed directories: {path}") # Parameter validation if offset < 0: raise ValueError("offset must be non-negative") if limit is not None and limit < 0: raise ValueError("limit must be non-negative") try: # Get file stats for metadata stats = await anyio.to_thread.run_sync(partial(abs_path.stat)) total_size = stats.st_size # Count total lines in file - we'll need this for context total_lines = 0 line_positions = [] # Store byte position of each line start async with await anyio.open_file(abs_path, "rb") as f: pos = 0 line_positions.append(pos) while True: line = await f.readline() if not line: break pos += len(line) total_lines += 1 # Always store the position of the start of each line # This ensures we have accurate line positions for all lines line_positions.append(pos) # Calculate the effective end offset if limit is specified end_offset = None if limit is not None: end_offset = offset + limit - 1 # Convert limit to inclusive end offset # Make sure we don't go beyond the file if offset >= total_lines: content = "" # Nothing to read else: # Adjust end_offset if it exceeds total lines if end_offset is None or end_offset >= total_lines: end_offset = total_lines - 1 # Determine byte positions to read start_pos = line_positions[offset] # Use 0-based offset directly # Calculate end position if end_offset >= len(line_positions) - 1: # If we're requesting the last line end_pos = total_size else: # Normal case - use the position of the line AFTER the end offset end_pos = line_positions[end_offset + 1] # Read the content async with await anyio.open_file(abs_path, "rb") as f: await f.seek(start_pos) content_bytes = await f.read(end_pos - start_pos) try: content = content_bytes.decode(encoding) except UnicodeDecodeError: raise ValueError(f"Cannot decode file as {encoding}") # Calculate the number of lines read if offset >= total_lines: lines_read = 0 elif end_offset is None: lines_read = total_lines - offset else: lines_read = min((end_offset - offset + 1), (total_lines - offset)) # Prepare metadata metadata = { "path": str(abs_path), "offset": offset, "limit": limit, "end_offset": end_offset, "total_lines": total_lines, "lines_read": lines_read, "total_size": total_size, "size_read": len(content), "encoding": encoding, } return content, metadata except FileNotFoundError: raise FileNotFoundError(f"File not found: {path}") except PermissionError: raise ValueError(f"Permission denied: {path}")
• mcp_filesystem/server.py:806-806 (registration)

  The @mcp.tool() decorator registers the read_file_lines function as an MCP tool.
  async def read_file_lines(
• mcp_filesystem/server.py:813-824 (schema)

  The docstring provides the schema definition for the tool's parameters and return value, used by FastMCP for input/output validation.
  """Read specific lines from a text file. Args: path: Path to the file offset: Line offset (0-based, starts at first line) limit: Maximum number of lines to read (None for all remaining) encoding: Text encoding (default: utf-8) ctx: MCP context Returns: File content and metadata """