from mcp.server.fastmcp import FastMCP
import os, re, math
from collections import Counter
from starlette.applications import Starlette
from starlette.routing import Route, Mount
from starlette.responses import JSONResponse
from starlette.requests import Request
import uvicorn
mcp = FastMCP(
"Music Personality",
instructions="""You are a riyaz (practice) journal companion for a Hindustani classical musician
who plays surbahar and sitar. The musician practices khayal in the Etawah (Imdadkhani) gharana
(gayaki ang, Vilayat Khan lineage) and dhrupad in the Dagarvani tradition.
When journaling riyaz: ask about the raga (aroha-avaroha, vadi-samvadi, pakad), explore
the style-specific elements (nom-tom for dhrupad, gayaki ang meend for khayal), discuss
what felt natural vs needs work, and use search_music_knowledge to verify raga grammar."""
)
KB_DIR = "/raga_kb"
def _tokenize(text: str) -> list:
return re.findall(r"[a-z]+", text.lower())
def _bm25(qt: list, dt: list, avg_dl: float) -> float:
k1, b = 1.5, 0.75
dl = len(dt)
tf = Counter(dt)
score = 0.0
for t in qt:
if t not in tf:
continue
freq = tf[t]
idf = math.log(2.0)
score += idf * (freq * (k1 + 1)) / (freq + k1 * (1 - b + b * dl / avg_dl))
return score
@mcp.tool()
def search_music_knowledge(query: str, max_results: int = 3) -> str:
if not os.path.exists(KB_DIR):
return "Knowledge base not available."
max_results = max(1, min(max_results, 5))
qt = _tokenize(query)
if not qt:
return "Please provide a search query."
docs = []
for fname in os.listdir(KB_DIR):
if not fname.endswith(".txt"):
continue
try:
with open(os.path.join(KB_DIR, fname), encoding="utf-8") as f:
content = f.read()
docs.append((fname[:-4].replace("_", " "), content, _tokenize(content)))
except Exception:
continue
if not docs:
return "Knowledge base is empty."
avg_dl = sum(len(d[2]) for d in docs) / len(docs)
scored = [(t, c, _bm25(qt, tok, avg_dl)) for t, c, tok in docs]
scored.sort(key=lambda x: x[2], reverse=True)
top = [(t, c) for t, c, s in scored[:max_results] if s > 0]
if not top:
return f"No results for: {query}"
parts = []
for title, content in top:
snippet = content[:1800]
ld = snippet.rfind(".")
if ld > 600:
snippet = snippet[:ld + 1]
parts.append(f"## {title}\n\n{snippet}")
return "\n\n---\n\n".join(parts)
@mcp.tool()
def get_raga_info(raga_name: str) -> str:
if not os.path.exists(KB_DIR):
return "Knowledge base not available."
target = raga_name.lower().replace(" ", "_")
candidates = []
for fname in os.listdir(KB_DIR):
if not fname.endswith(".txt"):
continue
base = fname[:-4].lower()
if base == target or base == target + "_(raga)":
candidates = [fname]
break
if target in base or base in target:
candidates.append(fname)
if not candidates:
return search_music_knowledge(raga_name, max_results=2)
with open(os.path.join(KB_DIR, candidates[0]), encoding="utf-8") as f:
return f.read()[:3000]
async def openapi_schema(request: Request):
return JSONResponse({
"openapi": "3.1.0",
"info": {"title": "Music Knowledge MCP", "version": "1.0.0",
"description": "Hindustani classical music knowledge base"},
"paths": {
"/search": {
"post": {
"operationId": "search_music_knowledge",
"summary": "Search the Hindustani music knowledge base",
"description": "Search for ragas, thaats, talas, gharanas and music theory",
"requestBody": {"required": True, "content": {"application/json": {"schema": {
"type": "object",
"required": ["query"],
"properties": {
"query": {"type": "string", "description": "Search query e.g. Yaman raga aroha"},
"max_results": {"type": "integer", "default": 3}
}
}}}},
"responses": {"200": {"description": "Search results", "content": {"application/json": {"schema": {"type": "object", "properties": {"result": {"type": "string"}}}}}}}
}
},
"/raga": {
"post": {
"operationId": "get_raga_info",
"summary": "Get detailed information about a specific raga",
"requestBody": {"required": True, "content": {"application/json": {"schema": {
"type": "object",
"required": ["raga_name"],
"properties": {"raga_name": {"type": "string", "description": "Raga name e.g. Yaman"}}
}}}},
"responses": {"200": {"description": "Raga information"}}
}
}
}
})
async def api_search(request: Request):
body = await request.json()
result = search_music_knowledge(body.get("query", ""), body.get("max_results", 3))
return JSONResponse({"result": result})
async def api_raga(request: Request):
body = await request.json()
result = get_raga_info(body.get("raga_name", ""))
return JSONResponse({"result": result})
MUSIC_PERSONALITY = """You are a riyaz (practice) journal companion for a Hindustani classical musician who plays surbahar and sitar.
The musician practices khayal in the Etawah (Imdadkhani) gharana (gayaki ang, Vilayat Khan lineage, lyrical meend, sur-centric approach) and dhrupad in the Dagarvani tradition (nom-tom alap, meditative nada, Dagar bani aesthetics).
When journaling riyaz:
- Ask about the raga: aroha-avaroha, vadi-samvadi, pakad, phrases explored
- Dhrupad: nom-tom syllables, sthayee/antara structure, which svaras felt anchored
- Khayal: gayaki ang phrasing, meend quality, whether the alap felt lyrical
- Discuss what felt natural vs needs work; suggest next session focus
- Use search_music_knowledge or get_raga_info to verify raga grammar
Use correct Hindustani music terminology. Be concise and musically specific."""
async def api_config(request: Request):
return JSONResponse({"features": {"system": True}})
async def system_prompt_endpoint(request: Request):
return JSONResponse({"prompt": MUSIC_PERSONALITY})
if __name__ == "__main__":
mcp_app = mcp.streamable_http_app()
app = Starlette(routes=[
Mount("/mcp", app=mcp_app),
Route("/openapi.json", openapi_schema),
Route("/api/config", api_config),
Route("/system", system_prompt_endpoint),
Route("/search", api_search, methods=["POST"]),
Route("/raga", api_raga, methods=["POST"]),
])
uvicorn.run(app, host="0.0.0.0", port=8765)