server.py•3.45 kB
import os
from fastapi import FastAPI, HTTPException, Request, Depends
from fastapi.responses import JSONResponse
from typing import Optional
from dotenv import load_dotenv
load_dotenv()
from . import tools
MCP_API_KEY = os.getenv("MCP_API_KEY", "").strip()
MCP_PORT = int(os.getenv("MCP_PORT", 3000))
app = FastAPI(title="Lidarr MCP API")
def check_api_key(request: Request):
if MCP_API_KEY:
auth = request.headers.get("authorization", "")
if auth != f"Bearer {MCP_API_KEY}":
raise HTTPException(status_code=401, detail="Unauthorized")
@app.exception_handler(Exception)
async def generic_exception_handler(request: Request, exc: Exception):
return JSONResponse(status_code=500, content={"error": str(exc)})
# Artists
@app.get("/mcp/artists")
def artists_list(dep: Optional[None] = Depends(check_api_key)):
return tools.tool_list_artists()
@app.get("/mcp/artists/{artist_id}")
def artists_get(artist_id: int, dep: Optional[None] = Depends(check_api_key)):
try:
return tools.tool_get_artist(artist_id)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/artists")
def artists_create(payload: dict, dep: Optional[None] = Depends(check_api_key)):
return tools.tool_create_artist(payload)
@app.put("/mcp/artists/{artist_id}")
def artists_update(artist_id: int, payload: dict, dep: Optional[None] = Depends(check_api_key)):
return tools.tool_update_artist(artist_id, payload)
@app.delete("/mcp/artists/{artist_id}")
def artists_delete(artist_id: int, dep: Optional[None] = Depends(check_api_key)):
return tools.tool_delete_artist(artist_id)
# Albums
@app.get("/mcp/albums")
def albums_list(artistId: int, dep: Optional[None] = Depends(check_api_key)):
return tools.tool_list_albums_by_artist(artistId)
@app.get("/mcp/albums/{album_id}")
def albums_get(album_id: int, dep: Optional[None] = Depends(check_api_key)):
return tools.tool_get_album(album_id)
@app.post("/mcp/albums")
def albums_create(payload: dict, dep: Optional[None] = Depends(check_api_key)):
return tools.tool_create_album(payload)
@app.put("/mcp/albums/{album_id}")
def albums_update(album_id: int, payload: dict, dep: Optional[None] = Depends(check_api_key)):
return tools.tool_update_album(album_id, payload)
@app.delete("/mcp/albums/{album_id}")
def albums_delete(album_id: int, dep: Optional[None] = Depends(check_api_key)):
return tools.tool_delete_album(album_id)
# Tracks
@app.get("/mcp/tracks")
def tracks_list(albumId: int, dep: Optional[None] = Depends(check_api_key)):
return tools.tool_list_tracks_by_album(albumId)
@app.get("/mcp/tracks/{track_id}")
def tracks_get(track_id: int, dep: Optional[None] = Depends(check_api_key)):
return tools.tool_get_track(track_id)
@app.post("/mcp/tracks")
def tracks_create(payload: dict, dep: Optional[None] = Depends(check_api_key)):
return tools.tool_create_track(payload)
@app.put("/mcp/tracks/{track_id}")
def tracks_update(track_id: int, payload: dict, dep: Optional[None] = Depends(check_api_key)):
return tools.tool_update_track(track_id, payload)
@app.delete("/mcp/tracks/{track_id}")
def tracks_delete(track_id: int, dep: Optional[None] = Depends(check_api_key)):
return tools.tool_delete_track(track_id)
if __name__ == "__main__":
import uvicorn
uvicorn.run("lidarr_mcp.server:app", host="0.0.0.0", port=MCP_PORT, reload=False)