[Python] How to play music(spotify) using livekit In python

I want to make my LiveKit agent be able to play music using a tool call.

This is my current code: agent.py

from dotenv import load_dotenv

from livekit import agents
from livekit.agents import AgentSession, Agent, RoomInputOptions
from livekit.plugins import (
noise_cancellation,
)
from livekit.plugins import google
from prompts import AGENT_INSTRUCTION, SESSION_INSTRUCTION
from tools import get_weather, search_web, play_music
load_dotenv()


class Assistant(Agent):
def __init__(self) -> None:
super().__init__(
instructions=AGENT_INSTRUCTION,
llm=google.beta.realtime.RealtimeModel(
voice="Charon",
temperature=0.8,
),
tools=[
get_weather,
search_web,
play_music,
],

)



async def entrypoint(ctx: agents.JobContext):
session = AgentSession(

)

await session.start(
room=ctx.room,
agent=Assistant(),
room_input_options=RoomInputOptions(
noise_cancellation=noise_cancellation.BVC(),
),
)

await ctx.connect()

await session.generate_reply(
instructions=SESSION_INSTRUCTION,
)


if __name__ == "__main__":
agents.cli.run_app(agents.WorkerOptions(entrypoint_fnc=entrypoint))


And this is tools.py:

import logging
from livekit.agents import function_tool, RunContext
import requests
from langchain_community.tools import DuckDuckGoSearchRun
import os
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from typing import Optional
from dotenv import load_dotenv
import base64
import webbrowser
import http.server
import socketserver
import urllib.parse

load_dotenv()

@function_tool()
async def get_weather(
context: RunContext, # type: ignore
city: str) -> str:
"""
Get the current weather for a given city.
"""
try:
response = requests.get(
f"https://wttr.in/{city}?format=3")
if response.status_code == 200:
logging.info(f"Weather for {city}: {response.text.strip()}")
return response.text.strip()
else:
logging.error(f"Failed to get weather for {city}: {response.status_code}")
return f"Could not retrieve weather for {city}."
except Exception as e:
logging.error(f"Error retrieving weather for {city}: {e}")
return f"An error occurred while retrieving weather for {city}."

@function_tool()
async def search_web(
context: RunContext, # type: ignore
query: str) -> str:
"""
Search the web using DuckDuckGo.
"""
try:
results = DuckDuckGoSearchRun().run(tool_input=query)
logging.info(f"Search results for '{query}': {results}")
return results
except Exception as e:
logging.error(f"Error searching the web for '{query}': {e}")
return f"An error occurred while searching the web for '{query}'."

@function_tool()
def play_music(song_name: str):
"""
Plays a song on Spotify.

Args:
song_name: The name of the song to play.
"""
try:
client_id = os.getenv("SPOTIPY_CLIENT_ID")
client_secret = os.getenv("SPOTIPY_CLIENT_SECRET")
redirect_uri = os.getenv("SPOTIPY_REDIRECT_URI")
scope = "user-modify-playback-state"

auth_url = f"https://accounts.spotify.com/authorize?client_id={client_id}&response_type=code&redirect_uri={redirect_uri}&scope={scope}"
webbrowser.open(auth_url)

class MyRequestHandler(http.server.SimpleHTTPRequestHandler):
def do_GET(self):
self.send_response(200)
self.send_header("Content-type", "text/html")
self.end_headers()
self.wfile.write(b"<html><h1>Authentication successful!</h1><p>You can close this window.</p></html>")
global code
code = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query).get('code', [None])[0]

with socketserver.TCPServer(("", 8888), MyRequestHandler) as httpd:
httpd.handle_request()

if not code:
print("Authentication failed.")
return

auth_header = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode()
token_url = "https://accounts.spotify.com/api/token"
token_data = {
"grant_type": "authorization_code",
"code": code,
"redirect_uri": redirect_uri
}
token_headers = {
"Authorization": f"Basic {auth_header}"
}
response = requests.post(token_url, data=token_data, headers=token_headers)
access_token = response.json().get("access_token")

if not access_token:
print("Failed to get access token.")
return

search_url = f"https://api.spotify.com/v1/search?q={song_name}&type=track&limit=1"
headers = {"Authorization": f"Bearer {access_token}"}
response = requests.get(search_url, headers=headers)
results = response.json()

if not results['tracks']['items']:
print(f"Song '{song_name}' not found.")
return

track_uri = results['tracks']['items'][0]['uri']
play_url = "https://api.spotify.com/v1/me/player/play"
play_data = {"uris": [track_uri]}
response = requests.put(play_url, headers=headers, json=play_data)

if response.status_code == 204:
print(f"Playing '{song_name}' on Spotify.")
else:
print(f"Failed to play song. Status code: {response.status_code}")

except Exception as e:
print(f"An error occurred: {e}")


I also have a prompt.py that has a custom prompt for my needs and custom behaviors that I have not provided it is in this format:

AGENT_INSTRUCTION ="""

This is the agent instruction prompt

"""


SESSION_INSTRUCTION = """

This is for behaviors in the session

"""


.env

LIVEKIT_URL=livekit_url_here
LIVEKIT_API_KEY=api_key_here
LIVEKIT_API_SECRET=api_secret_here

GOOGLE_API_KEY=google_api_key_here.


I hope all of this context helps

I have tried using the spotify API(Some of the vars are called SPOTIPY as I was trying that earlier).

Then once I ask the agent a browser tab opens saying to authenticate with my spotify account which I say yes to and ask it again then the tab opens again and it does not play the music.

There are no specific steps to reproduce it just copy my code and run them using python3 agent.py console and python3 agent.py dev and if you choose dev then you have to use the live kit playground and click connect and you can start interacting with it.

Continue reading...