Spaces:
Paused
Paused
| import re | |
| from markdownify import markdownify | |
| from smolagents import Tool | |
| class DownloadTaskAttachmentTool(Tool): | |
| name = "download_file" | |
| description = "Downloads the file attached to the task ID" | |
| inputs = {'task_id': {'type': 'string', 'description': 'The task id to download attachment from.'}} | |
| output_type = "string" | |
| def forward(self, task_id: str) -> str: | |
| """ | |
| Downloads a file associated with the given task ID. | |
| Returns the file path where the file is saved locally. | |
| """ | |
| file_url = f"{DEFAULT_API_URL}/files/{task_id}" | |
| local_file_path = f"downloads/{task_id}.file" | |
| print(f"Downloading file for task ID {task_id} from {file_url}...") | |
| try: | |
| response = requests.get(file_url, stream=True, timeout=15) | |
| response.raise_for_status() | |
| os.makedirs("downloads", exist_ok=True) | |
| with open(local_file_path, "wb") as file: | |
| for chunk in response.iter_content(chunk_size=8192): | |
| file.write(chunk) | |
| print(f"File downloaded successfully: {local_file_path}") | |
| return local_file_path | |
| except requests.exceptions.RequestException as e: | |
| print(f"Error downloading file for task {task_id}: {e}") | |
| raise | |
| def __init__(self, *args, **kwargs): | |
| self.is_initialized = False | |
| class VisitWebpageTool(Tool): | |
| name = "visit_webpage" | |
| description = "Visits a webpage at the given url and reads its content as a markdown string. Use this to browse webpages." | |
| inputs = {'url': {'type': 'string', 'description': 'The url of the webpage to visit.'}} | |
| output_type = "string" | |
| def forward(self, url: str) -> str: | |
| try: | |
| import requests | |
| from markdownify import markdownify | |
| from requests.exceptions import RequestException | |
| from smolagents.utils import truncate_content | |
| except ImportError as e: | |
| raise ImportError( | |
| "You must install packages `markdownify` and `requests` to run this tool: for instance run `pip install markdownify requests`." | |
| ) from e | |
| try: | |
| # Send a GET request to the URL with a 20-second timeout | |
| response = requests.get(url, timeout=20) | |
| response.raise_for_status() # Raise an exception for bad status codes | |
| # Convert the HTML content to Markdown | |
| markdown_content = markdownify(response.text).strip() | |
| # Remove multiple line breaks | |
| markdown_content = re.sub(r"\n{3,}", "\n\n", markdown_content) | |
| return truncate_content(markdown_content, 10000) | |
| except requests.exceptions.Timeout: | |
| return "The request timed out. Please try again later or check the URL." | |
| except RequestException as e: | |
| return f"Error fetching the webpage: {str(e)}" | |
| except Exception as e: | |
| return f"An unexpected error occurred: {str(e)}" | |
| def __init__(self, *args, **kwargs): | |
| self.is_initialized = False | |