from smolagents import CodeAgent,DuckDuckGoSearchTool, HfApiModel,load_tool,tool import datetime import requests import pytz import yaml import time from tools.final_answer import FinalAnswerTool import smtplib import random from email.mime.text import MIMEText from Gradio_UI import GradioUI VALID_CRYPTOS = ["sol", "bitcoin", "usd-coin", "polygon", "ethereum", "avalanche-2", "copw"] CURRENCY = "cop" QUOTE_EXPIRATION = 180 # 3 minutos en segundos active_quotes = {} # Guardará las cotizaciones activas con su tiempo de expiración # Below is an example of a tool that does nothing. Amaze us with your creativity ! @tool def my_custom_tool(arg1:str, arg2:int)-> str: #it's import to specify the return type #Keep this format for the description / args / args description but feel free to modify the tool """A tool that does nothing yet Args: arg1: the first argument arg2: the second argument """ return "What magic will you build ?" @tool def get_crypto_quote(crypto: str) -> str: """Fetches a price quote for a cryptocurrency in COP with a validity of 3 minutes. Args: crypto: The cryptocurrency symbol (e.g., 'SOL', 'BTC', 'USDC'). """ crypto = crypto.lower() if crypto not in ["sol", "btc", "usdc", "pol", "eth", "avax", "copw"]: return f"La criptomoneda {crypto.upper()} no está en la lista de cotización." # Mapear los nombres a CoinGecko coingecko_ids = { "sol": "solana", "btc": "bitcoin", "usdc": "usd-coin", "pol": "polkadot", "eth": "ethereum", "avax": "avalanche-2", "copw": "copw" # Si COPW no está en CoinGecko, habría que darle 1 a 1 con el cop } url = f"https://api.coingecko.com/api/v3/simple/price?ids={coingecko_ids[crypto]}&vs_currencies={CURRENCY}" response = requests.get(url) if response.status_code != 200: return "Error obteniendo el precio de la criptomoneda." data = response.json() price = data[coingecko_ids[crypto]][CURRENCY] quote_id = str(random.randint(100000, 999999)) # ID aleatorio para la cotización expiration_time = time.time() + QUOTE_EXPIRATION active_quotes[quote_id] = {"crypto": crypto, "price": price, "expires_at": expiration_time} return f"La cotización para {crypto.upper()} es {price:,.2f} COP. Código de cotización: {quote_id}. Válido por 3 minutos." @tool def confirm_or_cancel_quote(quote_id: str, action: str) -> str: """Allows the user to confirm or cancel a crypto quote within the given time frame. Args: quote_id: The unique identifier of the quote. action: The action to take ('aceptar' or 'cancelar'). """ if quote_id not in active_quotes: return "El código de cotización no es válido o ya expiró." quote = active_quotes[quote_id] current_time = time.time() if current_time > quote["expires_at"]: del active_quotes[quote_id] # Eliminar la cotización expirada return "La cotización ha expirado. Solicita una nueva." if action.lower() == "aceptar": return f"Has aceptado la cotización de {quote['crypto'].upper()} por {quote['price']:,.2f} COP. Procederemos a enviarte un correo de confirmación." elif action.lower() == "cancelar": del active_quotes[quote_id] return "Has cancelado la cotización." else: return "Acción inválida. Usa 'aceptar' o 'cancelar'." SMTP_SERVER = "smtp.gmail.com" # Puedes cambiarlo por tu servidor de correo SMTP_PORT = 587 SENDER_EMAIL = "tuemail@gmail.com" SENDER_PASSWORD = "tucontraseña" @tool def send_transaction_email(email: str, quote_id: str) -> str: """Sends an email confirmation with the transaction summary and an OTP code. Args: email: The recipient's email address. quote_id: The ID of the accepted quote. """ if quote_id not in active_quotes: return "El código de cotización no es válido o ya expiró." quote = active_quotes[quote_id] otp_code = random.randint(100000, 999999) subject = "Confirmación de tu Transacción de Criptomonedas" body = f""" ¡Hola! Has aceptado la cotización de {quote['crypto'].upper()} por {quote['price']:,.2f} COP. Tu código de confirmación (OTP) es: {otp_code} Este código es válido por los próximos 5 minutos. Gracias por usar nuestro servicio. """ msg = MIMEText(body) msg["Subject"] = subject msg["From"] = SENDER_EMAIL msg["To"] = email try: server = smtplib.SMTP(SMTP_SERVER, SMTP_PORT) server.starttls() server.login(SENDER_EMAIL, SENDER_PASSWORD) server.sendmail(SENDER_EMAIL, email, msg.as_string()) server.quit() return "Correo enviado con éxito. Revisa tu bandeja de entrada." except Exception as e: return f"Error al enviar el correo: {str(e)}" @tool def get_current_time_in_timezone(timezone: str) -> str: """A tool that fetches the current local time in a specified timezone. Args: timezone: A string representing a valid timezone (e.g., 'America/New_York'). """ try: # Create timezone object tz = pytz.timezone(timezone) # Get current time in that timezone local_time = datetime.datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S") return f"The current local time in {timezone} is: {local_time}" except Exception as e: return f"Error fetching time for timezone '{timezone}': {str(e)}" final_answer = FinalAnswerTool() # If the agent does not answer, the model is overloaded, please use another model or the following Hugging Face Endpoint that also contains qwen2.5 coder: # model_id='https://pflgm2locj2t89co.us-east-1.aws.endpoints.huggingface.cloud' model = HfApiModel( max_tokens=2096, temperature=0.4, model_id='Qwen/Qwen2.5-Coder-32B-Instruct',# it is possible that this model may be overloaded custom_role_conversions=None, ) # Import tool from Hub image_generation_tool = load_tool("agents-course/text-to-image", trust_remote_code=True) with open("prompts.yaml", 'r') as stream: prompt_templates = yaml.safe_load(stream) agent = CodeAgent( model=model, tools=[final_answer,get_crypto_quote,confirm_or_cancel_quote,send_transaction_email], ## add your tools here (don't remove final answer) max_steps=6, verbosity_level=1, grammar=None, planning_interval=None, name=None, description=None, prompt_templates=prompt_templates ) GradioUI(agent).launch()