fix: improve tool definitions for web_fetch and get_file_content_by_file_link

ex_app/lib/all_tools/files.py

@@ -2,13 +2,13 @@
 # SPDX-License-Identifier: AGPL-3.0-or-later
 import xml.etree.ElementTree as ET
 from urllib.parse import unquote
-import niquests
+
 from langchain_core.tools import tool
 from nc_py_api import AsyncNextcloudApp
 from nc_py_api.files.files_async import AsyncFilesAPI, FsNode
 
 from ex_app.lib.all_tools.lib.decorator import dangerous_tool, safe_tool
-from ex_app.lib.all_tools.lib.files import get_file_id_from_file_url
+from ex_app.lib.all_tools.lib.files import format_fs_node, get_file_content_from_int_link, get_file_id_from_file_url
 
 
 def _validate_path(path: str) -> str:
@@ -45,58 +45,14 @@ async def get_file_content(file_path: str):
 	@safe_tool
 	async def get_file_content_by_file_link(file_url: str):
 		"""
-		Get the content of a file given an internal Nextcloud link (e.g., https://host/index.php/f/12345)
-		:param file_url: the nextcloud-internal file URL
+		Get the content of a Nextcloud-internal file using its internal file link.
+		This is NOT for fetching arbitrary web URLs, use web_fetch for those.
+		Only use this tool when the URL points to a file stored in Nextcloud (e.g., https://cloud.example.com/index.php/f/12345 or https://cloud.example.com/f/12345).
+		:param file_url: a Nextcloud-internal file URL (must match the pattern https://<host>/f/<fileId> or https://<host>/index.php/f/<fileId>)
 		:return: text content of the file
 		"""
 
-		file_id = get_file_id_from_file_url(file_url)
-		# Generate a direct download link using the fileId
-		info = await nc.ocs('POST', '/ocs/v2.php/apps/dav/api/v1/direct', json={'fileId': file_id}, response_type='json')
-		download_url = info.get('ocs', {}).get('data', {}).get('url', None)
-
-		if not download_url:
-			raise Exception('Could not generate download URL from file id')
-
-		# Download the file from the direct download URL
-		response = await niquests.async_api.get(download_url)
-
-		return response.text
-
-	def __format_fs_node(fsnode: FsNode) -> dict:
-		# todo: permissions info
-		return {
-			'path': fsnode.user_path,
-			'file_id': fsnode.info.fileid,
-			'etag': fsnode.etag.replace('"', '').replace("'", ''),
-			'bytes': fsnode.info.size,
-			'creation_date': fsnode.info.creation_date.isoformat(),
-			'last_modified': fsnode.info.last_modified.isoformat(),
-			'mimetype': fsnode.info.mimetype,
-			'is_shared': fsnode.is_shared,
-			'is_favourite': fsnode.info.favorite,
-			'is_version': fsnode.info.is_version,
-			'trash_info': {
-				'in_trash': fsnode.info.in_trash,
-				**({
-					'trashbin_filename': fsnode.info.trashbin_filename,
-					'original_location': fsnode.info.trashbin_original_location,
-					'deletion_time': fsnode.info.trashbin_deletion_time,
-				} if fsnode.info.in_trash else {}),
-			},
-			'lock_info': {
-				'is_locked': fsnode.lock_info.is_locked,
-				**({
-					'owner': fsnode.lock_info.owner,
-					'owner_display_name': fsnode.lock_info.owner_display_name,
-					'type': fsnode.lock_info.type.name,
-					'creation_time': fsnode.lock_info.lock_creation_time,
-					'ttl': fsnode.lock_info.lock_ttl,
-					'locked_by_app': fsnode.lock_info.owner_editor,
-				} if fsnode.lock_info.is_locked else {}),
-			},
-		}
-
+		return await get_file_content_from_int_link(nc, file_url)
 
 	@tool
 	@safe_tool
@@ -112,7 +68,7 @@ async def get_file_tree(path: str = '/', include_metadata = False, depth: int =
 		files_handle = AsyncFilesAPI(nc._session)
 		fsnode_list = await files_handle.listdir(path, min(5, depth))
 		if include_metadata:
-			return [__format_fs_node(fsnode) for fsnode in fsnode_list]
+			return [format_fs_node(fsnode) for fsnode in fsnode_list]
 
 		return [fsnode.user_path for fsnode in fsnode_list]
 

ex_app/lib/all_tools/lib/files.py

@@ -2,12 +2,100 @@
 # SPDX-License-Identifier: AGPL-3.0-or-later
 import re
 
-def get_file_id_from_file_url(file_url) -> int:
-	# Define the regex pattern to capture only the digits
-	pattern = r"https?://[a-zA-Z-_.:0-9]+/(index\.php/)?f/(\d+)"
-	match = re.search(pattern, file_url)
+from nc_py_api import AsyncNextcloudApp
+from nc_py_api.files.files_async import AsyncFilesAPI, FsNode
 
+TEXT_LIKE_MIMETYPE_PARTS = ('text/', 'application/json', 'application/xml', 'application/xhtml', '+xml', '+json')
+MAX_FILE_SIZE = 100_000  # 100kB, approx. 25k tokens
+
+_FILE_ID_SUFFIX_RE = re.compile(r'/(index\.php/)?f/(\d+)/?$')
+
+
+def _strip_scheme(url: str) -> str:
+	"""Remove the http(s):// scheme from a URL for scheme-agnostic comparison."""
+	return re.sub(r'^https?://', '', url)
+
+
+def is_an_internal_file_link(nc: AsyncNextcloudApp, url: str) -> bool:
+	"""Check whether a URL is an internal file link for this Nextcloud instance (scheme-agnostic)."""
+	nc_host = _strip_scheme(nc.app_cfg.endpoint.rstrip('/'))
+	url_without_scheme = _strip_scheme(url)
+	return url_without_scheme.startswith(nc_host) and bool(_FILE_ID_SUFFIX_RE.search(url_without_scheme[len(nc_host):]))
+
+
+def get_file_id_from_file_url(file_url: str) -> int:
+	"""Extract the numeric file ID from a Nextcloud-internal file URL."""
+	match = _FILE_ID_SUFFIX_RE.search(file_url)
 	if match:
 		return int(match.group(2))
-	else:
-		raise Exception("Not a valid nextcloud file URL")
+	raise Exception("Not a valid nextcloud file URL")
+
+
+def format_fs_node(fsnode: FsNode) -> dict:
+	# todo: permissions info
+	return {
+		'path': fsnode.user_path,
+		'file_id': fsnode.info.fileid,
+		'etag': fsnode.etag.replace('"', '').replace("'", ''),
+		'bytes': fsnode.info.size,
+		'creation_date': fsnode.info.creation_date.isoformat(),
+		'last_modified': fsnode.info.last_modified.isoformat(),
+		'mimetype': fsnode.info.mimetype,
+		'is_shared': fsnode.is_shared,
+		'is_favourite': fsnode.info.favorite,
+		'is_version': fsnode.info.is_version,
+		'trash_info': {
+			'in_trash': fsnode.info.in_trash,
+			**({
+				'trashbin_filename': fsnode.info.trashbin_filename,
+				'original_location': fsnode.info.trashbin_original_location,
+				'deletion_time': fsnode.info.trashbin_deletion_time,
+			} if fsnode.info.in_trash else {}),
+		},
+		'lock_info': {
+			'is_locked': fsnode.lock_info.is_locked,
+			**({
+				'owner': fsnode.lock_info.owner,
+				'owner_display_name': fsnode.lock_info.owner_display_name,
+				'type': fsnode.lock_info.type.name,
+				'creation_time': fsnode.lock_info.lock_creation_time,
+				'ttl': fsnode.lock_info.lock_ttl,
+				'locked_by_app': fsnode.lock_info.owner_editor,
+			} if fsnode.lock_info.is_locked else {}),
+		},
+	}
+
+
+async def get_file_node(nc: AsyncNextcloudApp, file_id: int) -> FsNode:
+	files_handle = AsyncFilesAPI(nc._session)
+	node = await files_handle.by_id(file_id)
+	if not node:
+		raise RuntimeError(f'No file/folder found with id: {file_id}')
+	return node
+
+
+async def get_file_contents(nc: AsyncNextcloudApp, fsnode: FsNode) -> str:
+	"""
+	RuntimeError: just return the metadata to the model since the node is one of the following:
+		- a folder
+		- very large in size
+		- non-text mimetype
+	"""
+	if fsnode.is_dir:
+		raise RuntimeError('Folder found at the given file id, skipping download')
+	if fsnode.info.content_length > MAX_FILE_SIZE:
+		raise RuntimeError(f'File id {fsnode.info.fileid} is too large to download at {fsnode.info.content_length} bytes')
+	if not any(t in fsnode.info.mimetype for t in TEXT_LIKE_MIMETYPE_PARTS):
+		raise RuntimeError(f'File id {fsnode.info.fileid} is of content type {fsnode.info.mimetype} so cannot be displayed as text')
+	files_handle = AsyncFilesAPI(nc._session)
+	return (await files_handle.download(fsnode)).decode(encoding='utf-8', errors='ignore')
+
+
+async def get_file_content_from_int_link(nc: AsyncNextcloudApp, url: str) -> str:
+	file_id = get_file_id_from_file_url(url)
+	fsnode = await get_file_node(nc, file_id)
+
+	try:
+		return await get_file_contents(nc, fsnode)
+	except RuntimeError as e:
+		return f'Failed to download the file/folder: {e}.\nMore info about the node:{format_fs_node(fsnode)}'

ex_app/lib/all_tools/web.py

@@ -5,6 +5,12 @@
 from nc_py_api import AsyncNextcloudApp
 
 from ex_app.lib.all_tools.lib.decorator import safe_tool
+from ex_app.lib.all_tools.lib.files import (
+	MAX_FILE_SIZE,
+	TEXT_LIKE_MIMETYPE_PARTS,
+	get_file_content_from_int_link,
+	is_an_internal_file_link,
+)
 
 
 async def get_tools(nc: AsyncNextcloudApp):
@@ -13,12 +19,30 @@ async def get_tools(nc: AsyncNextcloudApp):
 	@safe_tool
 	async def web_fetch(url: str) -> str:
 		"""
-		Get the contents of a web page via HTTP
-		:param url: The HTTP URL to the web page (e.g. https://nextcloud.com/team/ )
-		:return: the web page content
+		Fetch the contents of an external web page via HTTP.
+		Use this for any URL on the public internet or intranet (e.g., https://www.eff.org/).
+		:param url: the HTTP(S) URL of the web page to fetch
+		:return: the raw web page content (HTML, JSON, etc.)
 		"""
-		res = await niquests.get(url)
-		return res.text()
+		# Detect Nextcloud-internal file links and fetch via the internal API, for the models that would still call this tool.
+		if is_an_internal_file_link(nc, url):
+			return await get_file_content_from_int_link(nc, url)
+
+		# Pre-flight HEAD request to check content type and size before downloading
+		head_res = await niquests.async_api.head(url, allow_redirects=True)
+
+		content_type = head_res.headers.get('Content-Type', '')
+		if not any(t in content_type for t in TEXT_LIKE_MIMETYPE_PARTS):
+			return f"(binary or unknown content detected: {content_type or 'no Content-Type header'}. Cannot display as text.)"
+
+		# download first 100kB of the file
+		res = await niquests.async_api.get(url, headers={'Range': f'bytes=0-{MAX_FILE_SIZE - 1}'})
+		text = res.text or "(empty content)"
+
+		if res.status_code == 206:
+			text += "\n\n(truncated: only the first 100 kB of content was fetched.)"
+
+		return text
 
 	return [
 		web_fetch,