TIDAL MCP Server

Instructions

Implementation Reference

• src/tidal_mcp/server.py:157-227 (handler)

  Implementation of the 'login' tool handler. Uses @mcp.tool decorator for automatic registration. Performs OAuth login via browser, persists session to file, checks authentication status.

  @mcp.tool(
      annotations={
          "title": "Authenticate with TIDAL",
          "readOnlyHint": False,
          "openWorldHint": True,
          "idempotentHint": False,
      }
  )
  async def login() -> AuthResult:
      """
      Authenticate with TIDAL using OAuth browser flow.
      Opens browser automatically for secure login.
      Session is persisted for future use.
      """
      if await ensure_authenticated():
          return AuthResult(
              status="success",
              message="Already authenticated with TIDAL",
              authenticated=True,
          )
  
      try:
          # Start OAuth device code flow
          login_obj, future = await anyio.to_thread.run_sync(session.login_oauth)
  
          auth_url = login_obj.verification_uri_complete
          if not auth_url.startswith("http"):
              auth_url = "https://" + auth_url
  
          # Try to open browser automatically
          try:
              await anyio.to_thread.run_sync(webbrowser.open, auth_url)
          except Exception:
              pass
  
          # Wait for user to complete authentication (polls TIDAL's server)
          # This blocks until auth completes or the link expires (~5 minutes)
          await anyio.to_thread.run_sync(future.result)
  
          # Check if login succeeded
          if await anyio.to_thread.run_sync(session.check_login):
              # Save session for future use
              session_data = {
                  "token_type": {"data": session.token_type or "Bearer"},
                  "session_id": {"data": session.session_id or ""},
                  "access_token": {"data": session.access_token},
                  "refresh_token": {"data": session.refresh_token},
                  "is_pkce": {"data": session.is_pkce},
              }
              async with await anyio.open_file(SESSION_FILE, "w") as f:
                  await f.write(json.dumps(session_data))
  
              return AuthResult(
                  status="success",
                  message="Successfully authenticated with TIDAL",
                  authenticated=True,
              )
          else:
              raise ToolError("Authentication failed - please try again")
      except ToolError:
          raise
      except Exception as e:
          error_msg = str(e)
          if "too long" in error_msg.lower() or "timeout" in error_msg.lower():
              raise ToolError(
                  f"Authentication timed out. Please authenticate using the helper script:\n\n"
                  f"  cd /home/ubuntu/code/personal/tidal/tidal-mcp && uv run python authenticate.py\n\n"
                  f"After authenticating, all MCP tools will work automatically."
              )
          raise ToolError(f"Authentication error: {error_msg}")

• src/tidal_mcp/models.py:123-129 (schema)

  Pydantic BaseModel defining the structured output schema for the login tool response.

  class AuthResult(BaseModel):
      """Result of authentication attempt."""
  
      status: str = Field(description="Operation status (success/error)")
      message: str = Field(description="Status message")
      authenticated: bool = Field(description="Whether authentication was successful")

• src/tidal_mcp/server.py:120-151 (helper)

  Helper function to check and load persisted authentication session. Used by login and all other authenticated tools.

  async def ensure_authenticated() -> bool:
      """
      Check if user is authenticated with TIDAL.
      Automatically loads persisted session if available.
      """
      if await anyio.Path(SESSION_FILE).exists():
          try:
              async with await anyio.open_file(SESSION_FILE, "r") as f:
                  content = await f.read()
                  data = json.loads(content)
  
              # Load OAuth session
              result = await anyio.to_thread.run_sync(
                  session.load_oauth_session,
                  data["token_type"]["data"],
                  data["access_token"]["data"],
                  data["refresh_token"]["data"],
                  None,  # expiry time
              )
  
              if result:
                  is_valid = await anyio.to_thread.run_sync(session.check_login)
                  if not is_valid:
                      await anyio.Path(SESSION_FILE).unlink()
                  return is_valid
              return False
          except Exception:
              await anyio.Path(SESSION_FILE).unlink()
              return False
  
      return await anyio.to_thread.run_sync(session.check_login)

• src/tidal_mcp/server.py:157-164 (registration)

  @mcp.tool decorator registers the login function as an MCP tool with custom annotations.

  @mcp.tool(
      annotations={
          "title": "Authenticate with TIDAL",
          "readOnlyHint": False,
          "openWorldHint": True,
          "idempotentHint": False,
      }
  )