Spotify을 활용한 팟캐스트 리서치 브리핑 에이전트
Spotify을 활용한 팟캐스트 리서치 브리핑 에이전트 (Podcast research briefing agent with Spotify)
Mistral 에이전트가 오케스트레이션하는 에이전트로, Spotify의 팟캐스트 카탈로그에서 에피소드를 검색하고, 웹 리서치로 보강하고, 순위가 매겨진 추천이 포함된 구조화된 브리핑을 얻는 쿡북이에요.
출처: 문서
본문
에이전트는 Agents API를 통해 세 가지 도구 유형을 조율해요:
| Tool type | What it provides |
|---|---|
| Spotify functions | 팟캐스트·에피소드 검색, 쇼 상세, 에피소드 상세 |
| Briefing function | LLM 기반 구조화된 브리핑 생성 |
| Web search (built-in) | 트랜스크립트, 게스트 프로필, 에피소드 요약 |
모든 도구는 인라인으로 function tool로 정의되므로, Mistral SDK와 spotipy 외에 외부 서버나 의존성이 없어요.
API status: 이 노트북은
client.beta.agents와client.beta.conversations를 사용해요. 이것들은 beta 엔드포인트라 변경될 수 있어요.
사전 요구 사항 (Prerequisites)
이 노트북을 완료하려면 다음이 필요해요:
- Python 3.11 이상
- Mistral 계정과 API 키
- Spotify Developer 자격 증명 (Client ID와 Client Secret)
Spotify 자격 증명 설정 (Setting up Spotify credentials)
- Spotify Developer Dashboard로 이동해서 Spotify 계정으로 로그인하세요. Web API를 사용하려면 Spotify Premium 구독이 필요해요.
- Create app을 클릭하세요.
- 폼을 작성하세요:
- App name: 아무 이름 (예: "Podcast Research Agent")
- App description: 아무 설명
- Redirect URI:
https://localhost:8080/callback을 입력하세요. (사용되진 않지만 필수 필드예요.) - Which API/SDKs are you planning to use?: Web API 선택
- 서비스 약관(terms of service) 박스에 체크하고 Save를 클릭하세요.
- 앱 대시보드에서 Settings를 클릭하세요.
- Client ID와 Client Secret을 복사하세요. ("View client secret"을 클릭해서 공개하세요.)
이 쿡북은 Client Credentials 인증 흐름을 사용하며, 이는 Spotify의 공개 카탈로그에 대한 읽기 전용 접근을 제공해요. 런타임에 사용자 로그인이나 OAuth 리다이렉트는 필요 없어요.
환경 설정 (Environment setup)
필요한 패키지를 설치하세요.
%pip install mistralai spotipy --quiet
필요한 모듈을 가져오고, API 키를 설정하고(환경 변수가 이미 설정되어 있지 않으면 보안 입력 프롬프트가 나타나요), Mistral 클라이언트를 초기화하세요.
import getpass
import json
import os
from IPython.display import display, Markdown
from mistralai.client import Mistral
from mistralai.client.models import (
FunctionCallEvent,
FunctionResultEntry,
MessageOutputEvent,
)
if not os.environ.get("MISTRAL_API_KEY"):
os.environ["MISTRAL_API_KEY"] = getpass.getpass("Mistral API key: ")
if not os.environ.get("SPOTIFY_CLIENT_ID"):
os.environ["SPOTIFY_CLIENT_ID"] = getpass.getpass("Spotify Client ID: ")
if not os.environ.get("SPOTIFY_CLIENT_SECRET"):
os.environ["SPOTIFY_CLIENT_SECRET"] = getpass.getpass("Spotify Client Secret: ")
client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
아키텍처 (Architecture)
에이전트는 에이전트에 직접 등록된 function tool을 사용해요. 에이전트가 도구를 호출하면, 스트리밍 루프가 해당 Python 함수를 실행하고 FunctionResultEntry로 결과를 다시 보내요.
┌─────────────────────┐
│ Mistral Agent │
│ (zai-glm-5-2) │
└──────┬──────┬────────┘
│ │
┌────────────────┘ └────────────────┐
│ │ │
┌─────────▼────────┐ ┌───▼──────────────┐ ┌─────▼──────────┐
│ Spotify functions │ │ Briefing function│ │ Web Search │
│ (spotipy client │ │ (mistral LLM │ │ (built-in) │
│ credentials) │ │ chat completion) │ │ │
└──────────────────┘ └──────────────────┘ └────────────────┘
- Spotify functions — Client Credentials 인증으로
spotipy를 통해 Spotify Web API를 감싸요. 팟캐스트 검색, 에피소드 검색, 상세 정보 조회 도구를 제공해요. - Briefing function —
zai-glm-5-2를 사용해서 수집된 팟캐스트 데이터와 웹 리서치로 구조화된 리서치 브리핑을 생성해요. - Web search — Mistral의 내장 웹 검색 도구로 트랜스크립트, 게스트 프로필, 에피소드 요약을 찾아 브리핑을 보강해요.
Step 1 — 도구 함수 정의 (Define tool functions)
Function tool을 사용하면 에이전트가 여러분의 코드를 호출할 수 있어요. 일반 Python 함수를 작성하면, 에이전트가 필요하다고 판단할 때 Agents API가 함수 이름과 인자가 포함된 FunctionCallEvent를 내보내요. 여러분의 코드는 함수를 로컬에서 실행하고 결과를 다시 보내므로, 에이전트가 여러분의 코드를 직접 실행하지 않아요.
도구는 여섯 개의 함수로 정의돼요. 다섯 개는 팟캐스트 카탈로그 쿼리를 위해 spotipy로 Spotify Web API를 감싸고, 하나는 수집된 데이터로 구조화된 브리핑을 생성하기 위해 Mistral Chat API를 호출해요. 각 함수는 JSON 문자열을 반환해서 에이전트가 결과를 파싱할 수 있게 해요.
import spotipy
from spotipy.oauth2 import SpotifyClientCredentials
sp = spotipy.Spotify(auth_manager=SpotifyClientCredentials(
client_id=os.environ["SPOTIFY_CLIENT_ID"],
client_secret=os.environ["SPOTIFY_CLIENT_SECRET"],
))
MODEL = "zai-glm-5-2"
BRIEFING_SYSTEM_PROMPT = """You are a research analyst. Given a topic, podcast data from
Spotify, and web research, produce a concise markdown briefing with:
- Executive summary (2-3 sentences)
- Ranked episode recommendations with relevance score, episode/show name, Spotify link, duration, release date, and a one-sentence summary
- Key themes across episodes
- Notable experts and guests
- Gaps and limitations
Use ONLY exact URLs from the input data. Never fabricate Spotify links."""
def _format_duration(ms: int) -> str:
"""Convert milliseconds to a human-readable duration string."""
minutes = ms // 60000
if minutes >= 60:
hours = minutes // 60
remaining = minutes % 60
return f"{hours}h {remaining}m"
return f"{minutes}m"
def search_podcasts(query: str, limit: int = 10) -> str:
"""Search for podcast shows on Spotify."""
try:
results = sp.search(q=query, type="show", limit=limit)
shows = []
for item in results.get("shows", {}).get("items", []):
if item is None:
continue
shows.append({
"id": item["id"],
"name": item["name"],
"publisher": item.get("publisher", "Unknown"),
"description": (item.get("description") or "")[:500],
"total_episodes": item.get("total_episodes", 0),
"url": item.get("external_urls", {}).get("spotify", ""),
})
return json.dumps(shows, indent=2)
except Exception as e:
return json.dumps({"error": str(e)})
def search_episodes(query: str, limit: int = 10) -> str:
"""Search for specific podcast episodes on Spotify."""
try:
results = sp.search(q=query, type="episode", limit=limit)
episodes = []
for item in results.get("episodes", {}).get("items", []):
if item is None:
continue
episodes.append({
"id": item["id"],
"name": item["name"],
"show_name": item.get("show", {}).get("name", "Unknown"),
"description": (item.get("description") or "")[:500],
"duration": _format_duration(item.get("duration_ms", 0)),
"release_date": item.get("release_date", "Unknown"),
"url": item.get("external_urls", {}).get("spotify", ""),
})
return json.dumps(episodes, indent=2)
except Exception as e:
return json.dumps({"error": str(e)})
def get_podcast_details(show_id: str) -> str:
"""Get full details for a specific podcast show."""
try:
show = sp.show(show_id)
return json.dumps({
"id": show["id"],
"name": show["name"],
"publisher": show.get("publisher", "Unknown"),
"description": (show.get("description") or "")[:1000],
"total_episodes": show.get("total_episodes", 0),
"languages": show.get("languages", []),
"url": show.get("external_urls", {}).get("spotify", ""),
}, indent=2)
except Exception as e:
return json.dumps({"error": str(e)})
def get_podcast_episodes(show_id: str, limit: int = 10) -> str:
"""Get episodes from a specific podcast show."""
try:
results = sp.show_episodes(show_id, limit=limit)
episodes = []
for item in results.get("items", []):
if item is None:
continue
episodes.append({
"id": item["id"],
"name": item["name"],
"description": (item.get("description") or "")[:500],
"duration": _format_duration(item.get("duration_ms", 0)),
"release_date": item.get("release_date", "Unknown"),
"url": item.get("external_urls", {}).get("spotify", ""),
})
return json.dumps(episodes, indent=2)
except Exception as e:
return json.dumps({"error": str(e)})
def get_episode_details(episode_id: str) -> str:
"""Get full details for a specific podcast episode."""
try:
episode = sp.episode(episode_id)
return json.dumps({
"id": episode["id"],
"name": episode["name"],
"show_name": episode.get("show", {}).get("name", "Unknown"),
"description": (episode.get("description") or "")[:2000],
"duration": _format_duration(episode.get("duration_ms", 0)),
"release_date": episode.get("release_date", "Unknown"),
"language": episode.get("language", "Unknown"),
"url": episode.get("external_urls", {}).get("spotify", ""),
}, indent=2)
except Exception as e:
return json.dumps({"error": str(e)})
def generate_research_briefing(topic: str, podcast_data: str, web_research: str) -> str:
"""Generate a structured research briefing from podcast data and web research."""
try:
response = client.chat.complete(
model=MODEL,
messages=[
{"role": "system", "content": BRIEFING_SYSTEM_PROMPT},
{"role": "user", "content": f"""Topic: {topic}
Podcast data:
{podcast_data}
Web research:
{web_research}"""},
],
)
return response.choices[0].message.content
except Exception as e:
return json.dumps({"error": str(e)})
Step 2 — 도구 스키마 정의 및 에이전트 생성 (Define tool schemas and create the agent)
에이전트가 어떤 함수를 호출할 수 있는지 알도록, 각 함수에 tool schema를 제공해요. 이는 JSON Schema 형식을 따르는 함수 이름, 설명, 파라미터 사양을 담은 dict예요. 에이전트는 이 스키마를 읽고 언제·어떻게 각 도구를 호출할지 결정해요.
또한 도구 이름을 Python 구현에 매핑하는 functions_mapping dict도 필요해요. 스트리밍 루프는 런타임에 호출을 디스패치하기 위해 이를 사용해요.
client.beta.agents.create_async로 에이전트를 만들고, tools 파라미터에 도구 스키마(내장 web_search 도구 포함)를 전달해요. instructions 필드는 에이전트가 멀티스텝 리서치 워크플로우에서 도구를 어떻게 사용할지 알려줘요.
def _tool(name: str, description: str, parameters: dict) -> dict:
"""Helper to build a function tool schema."""
return {"type": "function", "function": {"name": name, "description": description, "parameters": parameters}}
tools = [
_tool("search_podcasts", "Search for podcast shows on Spotify matching a topic or keyword.", {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Search query for finding podcast shows."},
"limit": {"type": "integer", "description": "Maximum number of results (default 10)."},
},
"required": ["query"],
}),
_tool("search_episodes", "Search for podcast episodes on Spotify matching a topic or keyword.", {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Search query for finding podcast episodes."},
"limit": {"type": "integer", "description": "Maximum number of results (default 10)."},
},
"required": ["query"],
}),
_tool("get_podcast_details", "Get full details for a specific podcast show by its Spotify ID.", {
"type": "object",
"properties": {
"show_id": {"type": "string", "description": "The Spotify show ID."},
},
"required": ["show_id"],
}),
_tool("get_podcast_episodes", "Get episodes from a specific podcast show.", {
"type": "object",
"properties": {
"show_id": {"type": "string", "description": "The Spotify show ID."},
"limit": {"type": "integer", "description": "Maximum number of episodes (default 10)."},
},
"required": ["show_id"],
}),
_tool("get_episode_details", "Get full details for a specific podcast episode by its Spotify ID.", {
"type": "object",
"properties": {
"episode_id": {"type": "string", "description": "The Spotify episode ID."},
},
"required": ["episode_id"],
}),
_tool("generate_research_briefing", "Generate a structured research briefing from podcast data and web research.", {
"type": "object",
"properties": {
"topic": {"type": "string", "description": "The research topic being investigated."},
"podcast_data": {"type": "string", "description": "JSON string of podcast and episode data from Spotify."},
"web_research": {"type": "string", "description": "Additional context gathered from web search."},
},
"required": ["topic", "podcast_data", "web_research"],
}),
{"type": "web_search"},
]
# Map tool names to Python functions for the streaming loop
functions_mapping = {
"search_podcasts": search_podcasts,
"search_episodes": search_episodes,
"get_podcast_details": get_podcast_details,
"get_podcast_episodes": get_podcast_episodes,
"get_episode_details": get_episode_details,
"generate_research_briefing": generate_research_briefing,
}
AGENT_INSTRUCTIONS = """Search Spotify for podcasts and episodes on the user's topic using
varied queries. Get details on the top results, use web search for additional context,
then pass the raw JSON data to generate_research_briefing. Never fabricate Spotify URLs."""
agent = await client.beta.agents.create_async(
model=MODEL,
name="podcast-research-agent",
instructions=AGENT_INSTRUCTIONS,
description="Podcast research briefing agent",
tools=tools,
)
print(f"Agent ready: {agent.name} (id={agent.id})")
Step 3 — 리서치 쿼리 실행 (Run a research query)
Conversations API는 에이전트와의 멀티턴 상호작용을 관리해요. conversations.start_stream_async를 호출해서 시작하고 이벤트 스트림을 받아요:
MessageOutputEvent— 에이전트의 텍스트 응답 조각으로, 토큰 단위로 스트리밍돼요.FunctionCallEvent— 에이전트가 function tool을 호출하려고 해요.tool_call_id, 함수name, JSON 문자열인arguments를 포함해요. 동일한 호출에 대해 여러 이벤트(스트리밍 인자 조각)가 오거나, 서로 다른 병렬 호출에 대해 여러 이벤트가 올 수 있어요.
run_research 함수는 전체 루프를 처리해요:
- 스트림에서 모든 호출을 수집하고,
tool_call_id로 인자 조각을 그룹화해요. functions_mapping으로 각 함수를 로컬에서 실행해요.conversations.append_stream_async로 결과를FunctionResultEntry객체 목록으로 다시 보내요.- 에이전트가 더 이상의 도구 호출 없이 끝날 때까지 반복해요.
async def run_research(query: str) -> str:
"""Run a podcast research query and return the briefing text."""
result = ""
conversation_id = None
response = await client.beta.conversations.start_stream_async(
agent_id=agent.id, inputs=query,
)
while True:
tool_calls = {}
async for event in response:
if not event.data:
continue
if conversation_id is None and hasattr(event.data, "conversation_id"):
conversation_id = event.data.conversation_id
match event.data:
case MessageOutputEvent():
if isinstance(event.data.content, str):
result += event.data.content
print(".", end="", flush=True)
case FunctionCallEvent():
call_id = event.data.tool_call_id
if call_id not in tool_calls:
tool_calls[call_id] = {"name": event.data.name, "arguments": ""}
print(f"\n[Tool call] {event.data.name}")
tool_calls[call_id]["arguments"] += event.data.arguments
if not tool_calls:
break
# Execute each function call and send results back
results = [
FunctionResultEntry(
tool_call_id=call_id,
result=functions_mapping[info["name"]](**json.loads(info["arguments"])),
)
for call_id, info in tool_calls.items()
]
response = await client.beta.conversations.append_stream_async(
conversation_id=conversation_id, inputs=results,
)
print(f"\n\nBriefing complete ({len(result)} chars)")
return result
QUERY = "Research podcasts about AI safety and alignment. Find episodes featuring leading researchers and recent developments."
briefing = await run_research(QUERY)
Step 4 — 브리핑 표시 (Display the briefing)
누적된 브리핑을 포맷된 마크다운으로 렌더링해요.
display(Markdown(briefing))
다른 주제 시도 (Try another topic)
run_research를 호출할 때마다 새 대화가 생성되므로, 에이전트는 이전 쿼리의 컨텍스트 없이 새로 시작해요. QUERY를 수정하고 셀을 실행하세요.
예시 주제:
- "Find podcast episodes covering climate technology and clean energy innovations"
- "Research podcast interviews with startup founders about lessons learned from building companies"
- "Podcasts about the history and future of space exploration"
QUERY = "Find podcast episodes covering climate technology and clean energy innovations"
new_briefing = await run_research(QUERY)
display(Markdown(new_briefing))
정리 (Cleanup)
에이전트는 삭제될 때까지 Mistral 서버에 유지돼요. 더 이상 사용할 계획이 없다면 작업이 끝날 때 에이전트를 삭제할 수 있어요. 에이전트와 연결된 대화도 함께 정리돼요.
await client.beta.agents.delete_async(agent_id=agent.id)
print(f"Agent deleted: {agent.id}")
요약 (Summary)
이 노트북은 Spotify를 검색하고, 웹 컨텍스트를 모으고, 구조화된 브리핑을 생성하는 팟캐스트 리서치 에이전트를 구축하는 방법을 보여줬어요.
만든 것 (What you built):
- 도구 스키마로 인라인 정의된 여섯 개의 function tool (Spotify 검색 + 브리핑 생성)
- 모든 도구와 내장 웹 검색을 오케스트레이션하는 팟캐스트 리서치 Mistral 에이전트
- 도구 호출을 로컬에서 실행하고 최종 브리핑을 마크다운으로 렌더링하는 스트리밍 파이프라인
사용한 Mistral 기능 (Mistral features used):
- Agents API (beta)
- 도구 실행을 위한
FunctionCallEvent/FunctionResultEntry를 사용한 Conversations API (beta) - 내장 웹 검색 도구
기타 서비스 (Other services):
- Spotify Web API —
spotipy를 통한 팟캐스트 카탈로그 검색
에이전트 구축에 대해 더 배우려면 Agents 문서를 참고하세요.