De DJ local a DJ con Spotify - tools externos y multi-agente

git clone https://github.com/hsaenzG/OpenSource-agents-demo.git
cd OpenSource-agents-demo
python3 -m venv .venv
source .venv/bin/activate
pip install 'strands-agents' spotipy python-dotenv

SPOTIFY_CLIENT_ID=TU-CLIENT-ID
SPOTIFY_CLIENT_SECRET=TU-CLIENT-SECRET

from strands import Agent, tool
from strands.models import BedrockModel
import json
import os
from dotenv import load_dotenv
import spotipy
from spotipy.oauth2 import SpotifyOAuth

load_dotenv()

sp = spotipy.Spotify(
    auth_manager=SpotifyOAuth(
        client_id=os.getenv("SPOTIFY_CLIENT_ID"),
        client_secret=os.getenv("SPOTIFY_CLIENT_SECRET"),
        redirect_uri="http://127.0.0.1:8000/callback",
        scope="playlist-modify-public,playlist-modify-private,user-library-read,user-top-read,user-modify-playback-state,user-read-playback-state",
    )
)

usuario = sp.current_user()
print(f"✅ Conectado a Spotify como: {usuario['display_name']}")

@tool
def buscar_en_spotify(query: str, limite: int = 10) -> str:
    """Busca canciones en Spotify por nombre, artista o género.
    SIEMPRE usa esta herramienta cuando el usuario pregunte por canciones o artistas.
    Los resultados son datos REALES y actualizados de Spotify.

    Args:
        query: Texto de búsqueda (ej: "Shakira", "rock alternativo", "Bad Bunny último")
        limite: Número máximo de resultados (default: 10)
    """
    resultados = sp.search(q=str(query), type="track", limit=min(limite, 10))
    tracks = resultados["tracks"]["items"]

    if not tracks:
        return f"No encontré canciones en Spotify para: {query}"

    canciones = []
    for t in tracks:
        canciones.append({
            "titulo": t["name"],
            "artista": t["artists"][0]["name"],
            "album": t["album"]["name"],
            "uri": t["uri"],
            "duracion_min": round(t["duration_ms"] / 60000, 1),
        })

    return json.dumps(canciones, ensure_ascii=False, indent=2)

@tool
def reproducir_cancion(nombre_cancion: str, artista: str = "") -> str:
    """Reproduce una canción en el dispositivo activo de Spotify del usuario.
    Busca la canción por nombre y la reproduce automáticamente.

    Requiere que Spotify esté abierto en algún dispositivo (celular, computadora, etc.).

    Args:
        nombre_cancion: Nombre de la canción a reproducir
        artista: Nombre del artista (opcional, ayuda a encontrar la correcta)
    """
    dispositivos = sp.devices()
    if not dispositivos["devices"]:
        return ("No hay dispositivos activos de Spotify. "
                "Abre Spotify en tu celular o computadora e intenta de nuevo.")

    # Buscar la canción
    query = f"track:{nombre_cancion}"
    if artista:
        query += f" artist:{artista}"
    resultados = sp.search(q=query, type="track", limit=5)
    tracks = resultados["tracks"]["items"]

    if not tracks:
        return f"No encontré '{nombre_cancion}' en Spotify."

    track = tracks[0]
    device_id = next(
        (d["id"] for d in dispositivos["devices"] if d["is_active"]),
        dispositivos["devices"][0]["id"]
    )

    sp.start_playback(device_id=device_id, uris=[track["uri"]])

    return json.dumps({
        "status": "reproduciendo",
        "cancion": track["name"],
        "artista": track["artists"][0]["name"],
        "mensaje": f"▶️ Reproduciendo: {track['name']} — {track['artists'][0]['name']}",
    }, ensure_ascii=False)

@tool
def crear_playlist_en_spotify(nombre: str, descripcion: str, canciones_uris: list) -> str:
    """Crea una playlist en la cuenta de Spotify del usuario con las canciones indicadas.

    Args:
        nombre: Nombre de la playlist (ej: "Viernes de Rock", "Cena Romántica")
        descripcion: Descripción breve de la playlist
        canciones_uris: Lista de URIs de Spotify o nombres de canciones
    """
    if not canciones_uris:
        return "No me diste canciones para agregar a la playlist."

    # Resolver URIs — si no es una URI válida, buscar la canción
    uris_validas = []
    for item in canciones_uris:
        item = str(item).strip()
        if item.startswith("spotify:track:"):
            uris_validas.append(item)
        else:
            r = sp.search(q=item, type="track", limit=1)
            tracks = r["tracks"]["items"]
            if tracks:
                uris_validas.append(tracks[0]["uri"])

    if not uris_validas:
        return "No pude encontrar ninguna de las canciones en Spotify."

    # Crear la playlist
    playlist = sp.user_playlist_create(
        user=sp.current_user()["id"],
        name=str(nombre),
        public=False,
        description=str(descripcion)
    )

    # Agregar canciones (en batches de 100, límite de la API)
    for i in range(0, len(uris_validas), 100):
        sp.playlist_add_items(playlist["id"], uris_validas[i:i + 100])

    return json.dumps({
        "status": "ok",
        "mensaje": f"Playlist '{nombre}' creada con {len(uris_validas)} canciones",
        "url": playlist["external_urls"]["spotify"],
    }, ensure_ascii=False)

@tool
def mis_top_artistas(periodo: str = "medium_term") -> str:
    """Obtiene los artistas más escuchados del usuario en Spotify.

    Args:
        periodo: "short_term" (último mes), "medium_term" (6 meses), "long_term" (siempre)
    """
    resultados = sp.current_user_top_artists(limit=10, time_range=periodo)
    artistas = []
    for a in resultados["items"]:
        artistas.append({
            "nombre": a["name"],
            "generos": a["genres"][:3],
            "popularidad": a["popularity"],
        })
    return json.dumps(artistas, ensure_ascii=False, indent=2)

@tool
def mis_top_canciones(periodo: str = "medium_term") -> str:
    """Obtiene las canciones más escuchadas del usuario en Spotify.

    Args:
        periodo: "short_term" (último mes), "medium_term" (6 meses), "long_term" (siempre)
    """
    resultados = sp.current_user_top_tracks(limit=20, time_range=periodo)
    canciones = []
    for t in resultados["items"]:
        canciones.append({
            "titulo": t["name"],
            "artista": t["artists"][0]["name"],
            "uri": t["uri"],
        })
    return json.dumps(canciones, ensure_ascii=False, indent=2)

modelo = BedrockModel(model_id="us.amazon.nova-pro-v1:0", region_name="us-east-1")

dj = Agent(
    model=modelo,
    system_prompt="""Eres un DJ personal conectado a Spotify. Controlas la música del usuario.

    REGLAS:
    1. SIEMPRE usa buscar_en_spotify antes de recomendar música.
    2. NUNCA inventes canciones, artistas o datos.
    3. Para reproducir: usa reproducir_cancion con el nombre.
    4. Para crear playlists: usa crear_playlist_en_spotify con las URIs.
    5. Basa TODAS tus respuestas en datos reales de las herramientas.

    Respondes en español, con onda y buen gusto musical.""",
    tools=[
        buscar_en_spotify,
        crear_playlist_en_spotify,
        reproducir_cancion,
        mis_top_artistas,
        mis_top_canciones,
    ],
)

# Conversación interactiva
while True:
    mensaje = input("🎵 Tú: ").strip()
    if mensaje.lower() in ("salir", "exit"):
        break
    print("\n🎧 DJ: ", end="", flush=True)
    dj(mensaje)
    print("\n")

@tool
def buscar_canciones(genero: str = "") -> str:
    """Busca canciones en la biblioteca local."""
    resultados = [c for c in BIBLIOTECA if genero.lower() in c["genero"].lower()]
    return json.dumps(resultados)

@tool
def buscar_en_spotify(query: str) -> str:
    """Busca canciones en Spotify."""
    resultados = sp.search(q=query, type="track", limit=10)
    return json.dumps(resultados["tracks"]["items"])

@tool
def consultar_dj_personal(mensaje: str) -> str:
    """Delega al DJ Personal: experto en gustos musicales y recomendaciones.

    Args:
        mensaje: El mensaje del usuario para el DJ Personal
    """
    respuesta = dj_personal(mensaje)
    return str(respuesta)

modelo = BedrockModel(model_id="us.amazon.nova-pro-v1:0", region_name="us-east-1")

dj_personal = Agent(
    model=modelo,
    system_prompt="""Eres un DJ personal experto. Conoces los gustos del usuario.
    SIEMPRE usa buscar_en_spotify antes de recomendar. NUNCA inventes datos.
    Puedes consultar mis_top_artistas y mis_top_canciones para conocer al usuario.""",
    tools=[buscar_en_spotify, crear_playlist_en_spotify, reproducir_cancion,
           mis_top_artistas, mis_top_canciones],
    callback_handler=None,  # Silenciar output
)

dj_eventos = Agent(
    model=modelo,
    system_prompt="""Eres un DJ profesional de eventos. Armas playlists para fiestas,
    bodas, cenas. Verificas que la duración cubra el evento completo.""",
    tools=[buscar_en_spotify, crear_playlist_en_spotify, reproducir_cancion,
           planificar_evento],
    callback_handler=None,
)

dj_emocional = Agent(
    model=modelo,
    system_prompt="""Eres un DJ empático especializado en emociones y música.
    Primero analizas la emoción, luego buscas música que la acompañe.
    Eres sensible y no juzgas.""",
    tools=[buscar_en_spotify, crear_playlist_en_spotify, reproducir_cancion,
           analizar_emocion],
    callback_handler=None,
)

@tool
def consultar_dj_personal(mensaje: str) -> str:
    """Delega al DJ Personal: experto en gustos musicales y recomendaciones.
    Úsalo cuando el usuario quiera recomendaciones, descubrir música nueva,
    o pida algo basado en sus gustos.

    Args:
        mensaje: El mensaje completo del usuario para el DJ Personal
    """
    respuesta = dj_personal(mensaje)
    return str(respuesta)

@tool
def consultar_dj_eventos(mensaje: str) -> str:
    """Delega al DJ de Eventos: experto en armar playlists para ocasiones específicas.
    Úsalo cuando el usuario mencione un evento, fiesta, boda, cena,
    o pida una playlist con duración específica.

    Args:
        mensaje: El mensaje completo del usuario para el DJ de Eventos
    """
    respuesta = dj_eventos(mensaje)
    return str(respuesta)

@tool
def consultar_dj_emocional(mensaje: str) -> str:
    """Delega al DJ Emocional: experto en música y estados de ánimo.
    Úsalo cuando el usuario exprese cómo se siente o quiera música
    para acompañar un estado de ánimo.

    Args:
        mensaje: El mensaje completo del usuario para el DJ Emocional
    """
    respuesta = dj_emocional(mensaje)
    return str(respuesta)

orquestador = Agent(
    model=modelo,
    system_prompt="""Eres el DJ principal. Tu trabajo es entender qué necesita el usuario
    y delegarlo al sub-agente especializado correcto.

    Tienes 3 DJs especializados:
    1. consultar_dj_personal — Recomendaciones basadas en gustos
    2. consultar_dj_eventos — Playlists para eventos con duración específica
    3. consultar_dj_emocional — Música para estados de ánimo

    REGLAS:
    - SIEMPRE delega al sub-agente apropiado.
    - Pasa el mensaje COMPLETO del usuario.
    - Si no estás seguro, usa consultar_dj_personal como default.
    - Presenta la respuesta del sub-agente de forma natural.""",
    tools=[consultar_dj_personal, consultar_dj_eventos, consultar_dj_emocional,
           reproducir_cancion, reproducir_playlist],
)

🎵 Tú: Estoy triste, ponme algo suave

🎧 DJ: [internamente: invoca consultar_dj_emocional("Estoy triste, ponme algo suave")]
       [DJ Emocional: invoca analizar_emocion("triste")]
       [DJ Emocional: invoca buscar_en_spotify("indie folk acoustic")]
       [DJ Emocional: invoca reproducir_cancion("Skinny Love", "Bon Iver")]

       Entiendo. Te puse "Skinny Love" de Bon Iver — indie folk suave,
       perfecto para este momento. Si quieres, puedo armar una playlist
       completa con ese mood.

# Capa 5 — Spotify
python capa5_spotify.py

# Capa 6 — Multi-agente
python capa6_multi_agente.py