File: //opt/PraisonAI/tools.py
# tools.py
from duckduckgo_search import DDGS
def search_tool(query: str) -> list:
"""
Perform a web search using DuckDuckGo and return relevant results.
Args:
query (str): The search query string to look up information about.
Returns:
list: A list of dictionaries containing search results with the following keys:
- title (str): The title of the search result
- url (str): The URL of the search result
- snippet (str): A brief excerpt or description of the search result
"""
try:
results = []
ddgs = DDGS()
for result in ddgs.text(keywords=query, max_results=10):
results.append({
"title": result.get("title", ""),
"url": result.get("href", ""),
"snippet": result.get("body", ""),
})
return results
except Exception as e:
print(f"Error during DuckDuckGo search: {e}")
return []
# # Define tools
# search_tool = {
# "type": "function",
# "function": {
# "name": "search_tool",
# "description": "Use this to perform search queries",
# "parameters": {
# "type": "object",
# "properties": {
# "query": {"type": "string"},
# },
# "required": ["query"],
# },
# },
# }
# from duckduckgo_search import DDGS
# from praisonai_tools import BaseTool
# class InternetSearchTool(BaseTool):
# name: str = "InternetSearchTool"
# description: str = "Search Internet for relevant information based on a query or latest news"
# def _run(self, query: str):
# ddgs = DDGS()
# results = ddgs.text(keywords=query, region='wt-wt', safesearch='moderate', max_results=5)
# return results
# from google.oauth2.credentials import Credentials
# from googleapiclient.discovery import build
# from google.auth.transport.requests import Request
# from google_auth_oauthlib.flow import Flow
# from google_auth_oauthlib.flow import InstalledAppFlow
# import os
# import json
# import webbrowser
# from http.server import HTTPServer, BaseHTTPRequestHandler
# from urllib.parse import urlparse, parse_qs
# import threading
# from datetime import datetime, timedelta
# import logging
# # Set up logging
# log_level = os.getenv('LOGLEVEL', 'INFO').upper()
# logging.basicConfig(level=log_level)
# logger = logging.getLogger(__name__)
# logger.setLevel(log_level)
# # Set up Google Calendar API
# SCOPES = ['https://www.googleapis.com/auth/calendar']
# def get_calendar_service():
# logger.debug("Getting calendar service")
# creds = None
# token_dir = os.path.join(os.path.expanduser('~'), '.praison')
# token_path = os.path.join(token_dir, 'token.json')
# credentials_path = os.path.join(os.getcwd(), 'credentials.json')
# if os.path.exists(token_path):
# creds = Credentials.from_authorized_user_file(token_path, SCOPES)
# logger.debug(f"Credentials loaded from {token_path}")
# if not creds or not creds.valid:
# if creds and creds.expired and creds.refresh_token:
# logger.debug(f"Refreshing credentials")
# creds.refresh(Request())
# else:
# logger.debug(f"Starting new OAuth 2.0 flow")
# flow = InstalledAppFlow.from_client_secrets_file(credentials_path, SCOPES)
# logger.debug(f"Credentials path: {credentials_path}")
# creds = flow.run_local_server(port=8090)
# logger.debug(f"Setting up flow from {credentials_path}")
# # creds = flow.run_local_server(port=8090) # Use run_local_server from InstalledAppFlow
# # Ensure the ~/.praison directory exists
# os.makedirs(os.path.dirname(token_path), exist_ok=True)
# logger.debug(f"Saving credentials to {token_path}")
# with open(token_path, 'w') as token:
# token.write(creds.to_json())
# logger.debug("Building calendar service")
# return build('calendar', 'v3', credentials=creds)
# check_calendar_def = {
# "name": "check_calendar",
# "description": "Check Google Calendar for events within a specified time range",
# "parameters": {
# "type": "object",
# "properties": {
# "start_time": {"type": "string", "description": "Start time in ISO format (e.g., '2023-04-20T09:00:00-07:00')"},
# "end_time": {"type": "string", "description": "End time in ISO format (e.g., '2023-04-20T17:00:00-07:00')"}
# },
# "required": ["start_time", "end_time"]
# }
# }
# async def check_calendar_handler(start_time, end_time):
# try:
# service = get_calendar_service()
# events_result = service.events().list(calendarId='primary', timeMin=start_time,
# timeMax=end_time, singleEvents=True,
# orderBy='startTime').execute()
# events = events_result.get('items', [])
# logger.debug(f"Found {len(events)} events in the calendar")
# logger.debug(f"Events: {events}")
# return json.dumps(events)
# except Exception as e:
# return {"error": str(e)}
# check_calendar = (check_calendar_def, check_calendar_handler)
# add_calendar_event_def = {
# "name": "add_calendar_event",
# "description": "Add a new event to Google Calendar",
# "parameters": {
# "type": "object",
# "properties": {
# "summary": {"type": "string", "description": "Event title"},
# "start_time": {"type": "string", "description": "Start time in ISO format"},
# "end_time": {"type": "string", "description": "End time in ISO format"},
# "description": {"type": "string", "description": "Event description"}
# },
# "required": ["summary", "start_time", "end_time"]
# }
# }
# async def add_calendar_event_handler(summary, start_time, end_time, description=""):
# try:
# service = get_calendar_service()
# event = {
# 'summary': summary,
# 'description': description,
# 'start': {'dateTime': start_time, 'timeZone': 'UTC'},
# 'end': {'dateTime': end_time, 'timeZone': 'UTC'},
# }
# event = service.events().insert(calendarId='primary', body=event).execute()
# logger.debug(f"Event added: {event}")
# return {"status": "success", "event_id": event['id']}
# except Exception as e:
# return {"error": str(e)}
# add_calendar_event = (add_calendar_event_def, add_calendar_event_handler)
# list_calendar_events_def = {
# "name": "list_calendar_events",
# "description": "List Google Calendar events for a specific date",
# "parameters": {
# "type": "object",
# "properties": {
# "date": {"type": "string", "description": "Date in YYYY-MM-DD format"}
# },
# "required": ["date"]
# }
# }
# async def list_calendar_events_handler(date):
# try:
# service = get_calendar_service()
# start_of_day = f"{date}T00:00:00Z"
# end_of_day = f"{date}T23:59:59Z"
# events_result = service.events().list(calendarId='primary', timeMin=start_of_day,
# timeMax=end_of_day, singleEvents=True,
# orderBy='startTime').execute()
# events = events_result.get('items', [])
# logger.debug(f"Found {len(events)} events in the calendar for {date}")
# logger.debug(f"Events: {events}")
# return json.dumps(events)
# except Exception as e:
# return {"error": str(e)}
# list_calendar_events = (list_calendar_events_def, list_calendar_events_handler)
# update_calendar_event_def = {
# "name": "update_calendar_event",
# "description": "Update an existing Google Calendar event",
# "parameters": {
# "type": "object",
# "properties": {
# "event_id": {"type": "string", "description": "ID of the event to update"},
# "summary": {"type": "string", "description": "New event title"},
# "start_time": {"type": "string", "description": "New start time in ISO format"},
# "end_time": {"type": "string", "description": "New end time in ISO format"},
# "description": {"type": "string", "description": "New event description"}
# },
# "required": ["event_id"]
# }
# }
# async def update_calendar_event_handler(event_id, summary=None, start_time=None, end_time=None, description=None):
# try:
# service = get_calendar_service()
# event = service.events().get(calendarId='primary', eventId=event_id).execute()
# if summary:
# event['summary'] = summary
# if description:
# event['description'] = description
# if start_time:
# event['start'] = {'dateTime': start_time, 'timeZone': 'UTC'}
# if end_time:
# event['end'] = {'dateTime': end_time, 'timeZone': 'UTC'}
# updated_event = service.events().update(calendarId='primary', eventId=event_id, body=event).execute()
# logger.debug(f"Event updated: {updated_event}")
# return {"status": "success", "updated_event": updated_event}
# except Exception as e:
# return {"error": str(e)}
# update_calendar_event = (update_calendar_event_def, update_calendar_event_handler)
# delete_calendar_event_def = {
# "name": "delete_calendar_event",
# "description": "Delete a Google Calendar event",
# "parameters": {
# "type": "object",
# "properties": {
# "event_id": {"type": "string", "description": "ID of the event to delete"}
# },
# "required": ["event_id"]
# }
# }
# async def delete_calendar_event_handler(event_id):
# try:
# service = get_calendar_service()
# service.events().delete(calendarId='primary', eventId=event_id).execute()
# logger.debug(f"Event deleted: {event_id}")
# return {"status": "success", "message": f"Event with ID {event_id} has been deleted"}
# except Exception as e:
# return {"error": str(e)}
# delete_calendar_event = (delete_calendar_event_def, delete_calendar_event_handler)
# tools = [
# check_calendar,
# add_calendar_event,
# list_calendar_events,
# update_calendar_event,
# delete_calendar_event,
# ]
# # Add this to the imports at the top of the file if not already present
# from datetime import datetime
# # Add this new tool definition and handler
# contract_expiry_checker_def = {
# "name": "check_contract_expiry",
# "description": "Check the expiry date of the contract",
# "parameters": {
# "type": "object",
# "properties": {},
# "required": []
# }
# }
# async def check_contract_expiry_handler():
# # In a real-world scenario, this would likely query a database or API
# # For this example, we'll return a fixed date
# expiry_date = datetime(2024, 12, 10).strftime("%d %B %Y")
# return {"expiry_date": expiry_date, "message": f"The contract expiry date is {expiry_date}."}
# check_contract_expiry = (contract_expiry_checker_def, check_contract_expiry_handler)
# # Add this to the tools list at the bottom of the file
# tools = [
# check_calendar,
# add_calendar_event,
# list_calendar_events,
# update_calendar_event,
# delete_calendar_event,
# check_contract_expiry, # Add this line
# ]